豪鬼メモ

一瞬千撃

DBサービスを作ろう その5 バッチ化とストリーム化による性能向上

前回の記事で述べた性能評価によって、gRPCによる操作が同一マシン上なら2万QPSのスループットで実行できることが分かった。これはデータベース自体の性能に比べると遅すぎるので、より高いスループットが出せるようにgRPCの使い方を工夫したくなる。今回は、バッチ化とストリーム化を行ってどの程度性能が向上できるかを確かめた。
f:id:fridaynight:20210903032228p:plain


一回のRPC呼び出しで複数レコードの検索や更新を行うことをバッチ化と呼ぶ。例えば、以下のメソッドは3つのレコードのデータを一気に取得する。

std::vector<std::string> keys = {"foo", "bar", "baz"};
std::map<std::string, std::string> records = dbm.GetMulti(keys);

以下のメソッドは、3つのレコードの格納と削除を順に行う。

std::map<std::string, std::string> records =
{{"foo", "hop"}, {"bar", "step"}, {"baz", "jump"}};
dbm.SetMulti(records);

std::vector<std::string> keys = {"foo", "bar", "baz"};
dbm.Remove(keys);

GetMultiのgRPCのメソッド定義は以下のようになる。

// GetMultiのリクエスト
message GetMultiRequest {
  // 操作対象のデータベースのインデックス
  int32 dbm_index = 1;
  // 取得するレコードのキーのリスト
  repeated bytes keys = 2;
}

// GetMultiのレスポンス
message GetMultiResponse {
  // ステータスコード
  StatusProto status = 1;
  // 該当レコードのkey-valueペアのリスト
  repeated BytesPair records = 2;
}

service DBMService {
  rpc GetMulti(GetMultiRequest) returns (GetMultiResponse);
}

アプリケーション側の工夫で複数操作のバッチ化ができるならば、最も効率が良い。個々のRPCのメッセージは大きくなるが、呼び出し回数が減るので、劇的に高速化する。とはいえ、物には順序があり、計算処理には依存関係があるので、うまいことバッチ化ができるユースケースはそんなに多くない。

次善策は、ストリーム化である。gRPCのストリームAPIでは、一回のRPC呼び出しの中で追加のデータの授受ができる。これはイテレータの実装で使ったのと同じ方法だ。ストリームオブジェクトを作ってから、それを介してGetやSetなどの操作を行うのだ。

std::unique_ptr<RemoteDBM::Stream> stream = dbm.GetStream();

stream.Set("foo", "hop");
stream.Set("bar", "step");

std::string value;
stream.Get("foo", &value);

stream.Remove("foo");

ストリームのgRPCのメソッド定義は以下のようになる。ストリームでないAPIのリクエストとレスポンスをそっくりそのままストリームに流し込んでいるだけだ。

// 元来のGetのリクエスト
message GetRequest {
  int32 dbm_index = 1;
  bytes key = 2;
  bool omit_value = 3;
}

// 元来のGetのレスポンス
message GetResponse {
  StatusProto status = 1;
  bytes value = 2;
}

// Streamメソッドのリクエスト
message StreamRequest {
  // 元来のメソッドのリクエストをoneofで埋め込む
  oneof request_oneof {
    EchoRequest echo_request = 1;
    GetRequest get_request = 2;
    SetRequest set_request = 3;
    RemoveRequest remove_request = 4;
  }
  // 更新系の場合、レスポンスを省略して高速化できる
  bool omit_response = 101;
}

// Streamメソッドのレスポンス
message StreamResponse {
  // リクエストの操作に対応したレスポンスを埋め込む
  oneof response_oneof {
    EchoResponse echo_response = 1;
    GetResponse get_response = 2;
    SetResponse set_response = 3;
    RemoveResponse remove_response = 4;
  }
}

service DBMService {
  rpc Stream(stream StreamRequest) returns (stream StreamResponse);
}

ストリームの利用で注意すべきは、個々のストリームはサーバ側の一つのスレッドを専有するということである。このことはサーバ側の実装を見るとわかりやすい。

grpc::Status StreamImpl(grpc::ServerContext* context,
                        grpc::ServerReaderWriterInterface<
                        tkrzw::StreamResponse, tkrzw::StreamRequest>* stream) {
  // ストリームが終わるまでループ
  while (true) {
    // キャンセルされたら抜ける
    if (context->IsCancelled()) {
      return grpc::Status(grpc::StatusCode::CANCELLED, "cancelled");
    }
    // 次のリクエストを読み込む
    tkrzw::StreamRequest request;
    if (!stream->Read(&request)) {
      break;
    }
    // リクエストの内容に応じた処理を行う
    tkrzw::StreamResponse response;
    switch (request.request_oneof_case()) {
      case tkrzw::StreamRequest::kGetRequest: {
        const grpc::Status status =
            Get(context, &request.get_request(), response.mutable_get_response());
        if (!status.ok()) {
          return status;
        }
        break;
      }
      ...
      default: {
        return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, "unknow request");
      }
    }
    // 結果を書き込む
    if (!request.omit_response() && !stream->Write(response)) {
      break;
    }
  }
  return grpc::Status::OK;
}

つまり、ストリームAPIは、RPCの初期化処理が呼び出される回数を減らしてオーバーヘッドを削減する効果がある反面、並列処理性能を下げてしまう。言い換えると、全体のスループットを上げる効果はあるが、個々のRPC呼び出しのレイテンシの偏差を上げてしまう副作用がある。よって、ストリームAPIが万能というわけではなく、適材適所で使い分けねばならない。

まとめると、バッチ化できるときはバッチAPIを使い、そうではなく、短時間に複数のデータベース操作を集中的に行う場合にはストリームAPIを使い、それでもない場合には通常のAPIを使うといいだろう。


さて、性能評価をしよう。前回と同様に、100万レコードを格納することを想定して、200万バケットに設定したハッシュデータベースを運用するデータベースサービスを立てる。ネットワーク層には、UNIXドメインソケットとIPv4を使い分ける。

# UNIXドメインソケットの場合
$ tkrzw_server --address "unix:$HOME/tkconn" "casket.tkh#num_buckets=2m"

# IPv4のTCPソケットの場合
$ tkrzw_server --address "127.0.0.1:1978" casket.tkh#num_buckets=2m"

クライアントは、以下のように実行する。実際には、実験内容に応じていろいろオプションを付ける。

$ tkrzw_dbm_remote_perf sequence --address "unix:$HOME/tkconn" --iter 1000000 --threads 1
$ tkrzw_dbm_remote_perf sequence --address "127.0.0.1" --iter 1000000 --threads 1

以下の操作を測定する。操作は合計100万回行う。クライアント側のスレッドは1、2、4とする。

  • normal-Echo : 通常APIのEcho
  • normal-Set : 通常APIのSet
  • normal-Get : 通常APIのGet
  • normal-Remove : 通常APIのGet
  • multi-Echo-10 : Echoを10メッセージ分まとめて実行
  • multi-Set-10 : バッチAPIのSetMulti。10レコードをまとめて実行
  • multi-Get-10 : バッチAPIのGetMulti。10レコードをまとめて実行
  • multi-Remove-10 : バッチAPIのRemoveMulti。10レコードをまとめて実行
  • multi-Echo-100 : Echoを100メッセージ分まとめて実行
  • multi-Set-100 : バッチAPIのSetMulti。100レコードをまとめて実行
  • multi-Get-100 : バッチAPIのGetMulti。100レコードをまとめて実行
  • multi-Remove-100 : バッチAPIのRemoveMulti。100レコードをまとめて実行
  • Stream-Echo : ストリームAPIのEcho
  • Stream-Set : ストリームAPIのSet
  • Stream-Get : ストリームAPIのGet
  • Stream-Remove : ストリームAPIのRemove
  • Stream-Set-NR : ストリームAPIのSet。レスポンスの確認を省略
  • Stream-Remove-NR : ストリームAPIのRemove。レスポンスの確認を省略

結果は以下のようになる。数値の単位はQPS(クエリ毎秒)である。

UNIX 1 thread UNIX 2 thread UNIX 4 thread IPv4 1thread IPv4 2thread IPv4 4thread
normal-Echo 20,347 28,197 25,292 16,887 24,168 26,510
normal-Set 19,013 21,791 21,563 15,523 19,143 21,479
normal-Get 18,356 20,762 24,483 16,679 20,678 21,714
normal-Remove 17,557 19,272 22,204 16,469 19,636 22,102
multi-Echo-10 180,145 253,989 308,470 155,716 232,544 258,576
multi-Set-10 140,446 192,152 260,460 123,170 188,309 226,687
multi-Get-10 137,117 193,273 237,894 120,738 178,721 215,959
multi-Remove-10 155,806 166,968 258,288 138,660 196,866 255,105
multi-Echo-100 1,439,564 1,973,220 2,720,126 1,000,875 1,827,947 2,473,265
multi-Set-100 458,714 746,803 1,188,727 323,048 761,758 1,036,163
multi-Get-100 493,673 769,554 1,067,237 286,206 732,079 1,066,825
multi-Remove-100 703,227 987,542 1,310,496 544,677 1,019,257 1,331,260
Stream-Echo 26,849 32,829 30,449 22,452 29,155 25,800
Stream-Set 24,921 26,524 23,256 21,058 24,131 20,255
Stream-Get 22,635 25,337 24,260 20,445 23,463 22,550
Stream-Remove 22,792 26,432 22,796 20,693 23,231 20,645
Stream-Set-NR 170,516 122,711 100,274 104,200 67,788 67,992
Stream-Remove-NR 183,737 122,710 93,004 105,652 61,283 61,939

まず、1列目の、UNIXドメインスレッドの1スレッドの結果に着目しよう。一目瞭然なのは、バッチ化最強ということだ。10回の操作をバッチ化できれば、RPCの呼び出し回数を1/10にできるので、それだけ高速化する。当然といえば当然だ。Echoとその他のメソッドの差がデータベース操作のオーバーヘッドとみなせるが、バッチ化100個の4スレッドになると始めてデータベースの性能が目に見えるようになる。Echo272万QPSに対してSet118万QPSになり、やっとデータベースがボトルネックになる。

2列目と3列目も見て、スレッド数とスループットの関係を考察する。通常のAPIでは、2スレッドまではスループット向上の効果があるが、4スレッドにしても意味がないということがわかる。一方で、バッチ化した場合、スレッド数に応じてスループットが顕著に上がる。このことから考えると、gRPCのメソッド呼び出しにかかるオーバーヘッドはマルチスレッドによる並列処理では改善しないらしい。

次に目をつけるべきは、ストリームAPIでレスポンスの確認を省略した場合だ。17万QPSでSetでき、18万QPSでRemoveできる。レスポンスの確認をしないGetは意味がないので測定していない。レスポンスを確認しないとなぜここまで高速化するかというと、ストリームが単方向になるからだ。ReadとWriteを繰り返す双方向ストリームだと、epollなどで入出力のモードを切り替える必要があるが、単方向だとその必要がない。レスポンスの確認省略で最も高いスループットが出るのが1スレッドだというのも興味深いところだ。サーバ側でリクエストを直列化する処理にボトルネックがある気がするが、あくまで推測だ。

最後に、UNIXドメインソケットとIPv4を比較しよう。当然ながらUNIXドメインソケットの方が早いのだが、スレッド数を増やすと差が縮まるところが興味深い。単純なデータ転送速度ではUNIXドメインソケットに優位性があるのだろうが、スレッドを切り替えながら細かいデータの授受を行うような場合だと、TCP/IPと大して変わらない性能になってしまうらしい。


まとめ。Tkrzw-RPCのAPIにバッチ化APIとストリーム化APIを加えて、ユースケースによって最適なモードを選べるようにした。予想通り、バッチ化は劇的な効果があり、100万QPSを超えるスループットが出せるようになる。とはいえバッチ化は常に適用できるわけではない。更新系に関してはストリームAPIのレスポンス省略モードが便利で、その場合、10万QPSとかで操作できるようになる。参照系に関してはバッチ化以外の技はない。次回は、非同期化でスループットが上がるかどうか検討する。