cybozu.com の MQ と性能問題 - Cybozu Inside Out | サイボウズエンジニアのブログ

いよいよ「サイボウズ・アドベントカレンダー2012」の始まりです。記事一覧も作成しましたので、ご活用ください。

こんにちは。CyDE-C チームの青木です。

CyDE-C チームとは cybozu.com のミドルウェア層を担当するチームで、アプリケーション層のチームに MQ や全文検索サービス、BLOBサービス、Slash と呼ばれるユーザー管理機能を提供しています。

今回は CyDE-C が提供する MQ のパフォーマンスチューニングについて紹介します。

MQ とは Message Queue の略で、cybozu.com ではタスクの非同期実行の要として使われています。この仕組みはジョブキューやタスクキューとも呼ばれますが、この記事では MQ で統一することにします。

現行の MQ について

cybozu.com の MQ は、キューに MySQL 、ワーカーは Java 製プログラムという構成になっています。ワーカーはキューを毎秒ポーリングして、ジョブがあれば処理をする、という仕組みになってます。また、この MQ の特徴として、ワーカーとキューの個数の関係は 1:1 ではなく 1:N になっています。これは、各ユーザーが使用するアプリごとにキューが存在しているためです。各キューにはそれぞれ対応するポーリングスレッドがつきます。従ってキューの個数分のポーリングスレッドが存在し、かつ各タスクを実行する際には更に新規にスレッドを立てて実行しています。

これは一見不要にリソースを消費しているように見えますが、この仕組みがあることでプロテクションドメイン(データが混線して他のユーザーに見えてしまうのを防ぐこと)がより強固になります。あるキューに対応する処理スレッドが、他のユーザーのキュー内のデータを処理してしまうことはありません。

性能問題

しばらくはこの仕組みでも問題なかったのですが、ユーザーが増えてきたことにより性能問題が出てきました。監視するキューが増加するにつれてポーリングスレッドが増え、ワーカーと MySQL の負荷が高くなりました。ワーカーのスレッド数は常時 1000 を超え、2000 を超えているワーカーも出てきました。また、スレッド数が増えるに従って CPU 使用率も増加しました。

このワーカーの問題は、各キューに対してポーリングを行なっていることです。ポーリングを廃止し、キューにジョブが挿入された時だけ処理スレッドが立ち上がり、そうでない時はスレッドは立ち上がらず、常に最小のスレッド数で済むように修正しましょう。

解決手段

では、どのようにしてポーリングを廃止すれば良いでしょうか。

今回は下記の三点をポイントに実装を進めることにしました。

ジョブが挿入された時、自発的にキューがワーカーにジョブが挿入された事を通知する
ワーカーは通知を受け取ったら処理スレッドを立ち上げるようにする
処理するジョブが無くなったら処理スレッドを終了する

さて、ワーカーはお手製の Java プログラムなので「通知を受け取ったら処理スレッドを立ち上げるようにする」、「処理するジョブが無くなったら処理スレッドを終了する」というように変更するのは簡単です。しかし、キューにジョブが挿入された時にキューがワーカーに通知してくれるようにするにはどうすれば良いのでしょうか？

MySQL の Trigger をキューに仕掛け、レコード(ジョブ)が挿入されたら Trigger が発動するようにします。その時に実行されるクエリ内でワーカーに通知する処理を行えば良いのですが、通常の SQL だけではそのような処理を記述することはできません。

ここで MySQL の Trigger + ユーザー定義関数 (UDF, User Defined Function) を使います。UDF は MySQL plugin の仕組みの1つで、SQL から利用可能な関数を自由に定義・追加することができます。この仕組みを利用して、ワーカーに通知を行う関数を作成しましょう。

UDF 実装

UDF は C/C++ で実装します。ここでは、notify という名前の関数を SQL 上から使えるようにしましょう。

最小の UDF は下記のようなコードで実現できます。

#include <mysql.h>
#include <m_string.h>

extern "C" {
  my_bool notify_init(UDF_INIT *initid, UDF_ARGS *args, char *message);
  my_ulonglong notify(UDF_INIT *initid, UDF_ARGS *args, char *is_null, char *error);
  void notify_deinit(UDF_INIT *initid);
}

// notify関数実行前に呼ばれる
my_bool notify_init(UDF_INIT *initid, UDF_ARGS *args, char *message){
    return 0;
}

// notify関数呼び出し終了後に呼ばれる
void notify_deinit(UDF_INIT *initid){
}

// notify関数本体
my_ulonglong notify(UDF_INIT *initid, UDF_ARGS *args, char *is_null, char *error){
    return 0;
}

このコードをコンパイル後、MySQLのプラグインディレクトリにバイナリを入れ、下記クエリを流します。

mysql> CREATE FUNCTION notify RETURNS int SONAME 'notify.so';
Query OK, 0 rows affected (0.00 sec)

すると、クエリ内で notify 関数が使えるようになります。

mysql> SELECT notify();
+----------+
| notify() |
+----------+
|        0 |
+----------+
1 row in set (0.00 sec)

あとは上記 C++ コードに肉付けして、引数を元にワーカーに通知を行うコードを書き、キューにジョブが積まれたら通知を行うようトリガを仕掛ければ OK です。

DELIMITER $$
CREATE TRIGGER notify_to_worker BEFORE INSERT ON _queue
  FOR EACH ROW
  BEGIN
    SET @ret = notify( DATABASE() );
  END$$
DELIMITER ;

通知方法

MySQL からワーカーへの通知はどのように行うのが良いでしょうか？まず思いついたのは HTTP を用いる方法ですが、毎回 TCP でコネクションを確立するのは下記2つの理由で考えものです。

ハンドシェイクのコストが大きい
クローズ後も TIME_WAIT 状態でポートを数十秒程度専有する -> 大量のジョブが一斉に積まれた際にはポートを埋め尽くす可能性がある

これらの理由から、ジョブが挿入された際のワーカーへの通知は UDP で行うことにします。

しかし UDP を用いると高速にパケットを送信できる反面、信頼性が失われます。通知パケットが消失してしまったらワーカーは何もしてくれないので、これを救済する仕組みを入れなければなりません。

この仕組みは MySQL の Event を用います。あまり知られていませんが、MySQL は Event と呼ばれるスケジューラの機能を持ってます。この機能を利用して数分に一度の頻度でワーカーに通知する処理をいれてやります。この時は信頼性が欲しいので UDP ではなく TCP を使います。つまり今回実装した UDF は TCP でワーカーに通知する機能と UDP でワーカーに通知する機能の2つを持っているわけです。

以下が定期的に通知を行うクエリです。

SET @INTERVAL_MINUTE = 5;
DELIMITER $$
CREATE EVENT scheduled_notifier
  ON SCHEDULE EVERY @INTERVAL_MINUTE MINUTE STARTS CURRENT_TIMESTAMP
  DO
  BEGIN
    SET @ret = notify( DATABASE(), 1 ); // TCP を用いて通知を行う
  END$$
DELIMITER ;

これで定期的に TCP で通知を行なってくれるようになったため、通常の通知は UDP で行なって問題無いようになりました。

しかし、実は上記クエリにも改善点があります。このクエリの Event はお客様が使用しているアプリの数だけ登録するのですが、そうなるとインターバル5分を超えたら一気に Event が同時発動してしまいます。それを防ぐため各アプリごとに Event の発動時間を分散させるようにしましょう。

以下のクエリが Event の発動時間を適度に分散させる版です。 INTERVAL FLOOR 〜〜のあたりで RAND 関数を用い、Event 発動時間を適度に分散させています。

SET @INTERVAL_MINUTE = 5;
DELIMITER $$
CREATE EVENT scheduled_notifier
  ON SCHEDULE EVERY @INTERVAL_MINUTE MINUTE
    STARTS CURRENT_TIMESTAMP + INTERVAL FLOOR(RAND()*@INTERVAL_MINUTE*60) SECOND
  DO
  BEGIN
    SET @ret = notify( DATABASE(), 1 );
  END$$
DELIMITER ;