MySQL の接続圧縮制御について - それが僕には楽しかったんです。

はじめに
- 前提
接続圧縮制御 is 何
接続圧縮制御と C API
どのように実現しているのか
余談
おわりに

はじめに

どうも、ブログを更新せず数年が経過していたことに驚きが隠せないけんつです。流石にネタはいくつかあるのでぼちぼち書いていこうと思います。というわけで今回は MySQL の接続圧縮制御というやつについて割と調べたのでそれについてです。

前提

今回も実装の話になると思いますが、対象のバージョンは 8.4.7 とします。

接続圧縮制御 is 何

とは言っても接続圧縮制御(Connection Compression Control)という名称自体あまり馴染みがないと思われるわけですが、要はこれです。
dev.mysql.com

ざっくり何かというと、特定のクライアントから MySQL に対して通信を行う場合に、そのパケットを圧縮して実際に送信するデータの容量を減らすことができるというナイスなやつ。現段階では基本的に zstd, zlib, uncompressed を選択できる。
とはいえ、勝手になんでもかんでも圧縮されると MySQL 側で受け取った時に圧縮データを展開するために CPU をバカ喰いしかねないので、サーバー側で許可している圧縮アルゴリズムとクライアントが使用する圧縮アルゴリズムをネゴシエーションすると言ったナイスな仕組みもあります。これは例えば公式のコマンド群(mysql, mysqldump, mysqlbinlog, ...)では大抵利用可能だが、一点注意が必要でこの接続圧縮制御というものにはレガシーなものが存在する。どうしてレガシーなのか非常に謎でどうしてそのような分岐が入ったのかも謎であるがそちらを使うと zlib を使って圧縮するか圧縮しないかの２択になるのである。
さらに罠なのはこれは接続圧縮制御を使う側の話であるが事情を知らなければ --compress という一見それらしいフラグがレガシーな接続圧縮制御のクライアントオプションであり、--compression-algorithms= がナウい方の接続圧縮制御のクライアントオプションとして宣言されているところである。
現時点では --compression-algorithms を使うのが良いだろう。

接続圧縮制御と C API

何で多くの公式コマンドで利用かというと、そもそもにこの接続圧縮制御というものは libmysqlclient によって提供される C API で汎用的に利用することが可能なのである。

どうやって使うかというと、コネクションに関わるあれこれは mysql_options 関数によって制御できるので、その関数に MYSQL_OPT_COMPRESSION_ALGORITHMS フラグと使いたい圧縮アルゴリズムを文字列で渡してやれば良い。
もし zstd で圧縮したい場合は圧縮レベルを MYSQL_OPT_ZSTD_COMPRESSION_LEVEL で同様に指定することもできる。
dev.mysql.com

例えば以下のような形になるはずである。

// zstd or zlib で圧縮したい場合
mysql_options(mysql, MYSQL_OPT_COMPRESSION_ALGORITHMS, "zstd,zlib");

というわけで案外簡単に何でも圧縮できるようになる。

と、それだけの話であるならばドキュメントで十分なのでわざわざブログにまとめることもないわけで本題はここから。

どのように実現しているのか

前述の mysql_options が MYSQL_OPT_COMPRESSION_ALGORITHMS と共に呼ばれると与えられたアルゴリズムのリストを探索しながらフラグを更新していく。
またこの時共通して行われるのは mysql->options.compress を true にすることである。

 case MYSQL_OPT_COMPRESSION_ALGORITHMS: {
      std::string compress_option(static_cast<const char *>(arg));
      std::vector<std::string> list;
      parse_compression_algorithms_list(compress_option, list);
      ENSURE_EXTENSIONS_PRESENT(&mysql->options);
      mysql->options.extension->connection_compressed = true;
      mysql->options.client_flag &=
          ~(CLIENT_COMPRESS | CLIENT_ZSTD_COMPRESSION_ALGORITHM);
      mysql->options.compress = false;
      auto it = list.begin();
      unsigned int cnt = 0;
      while (it != list.end() && cnt < COMPRESSION_ALGORITHM_COUNT_MAX) {
        std::string value = *it;
        switch (get_compression_algorithm(value)) {
          case enum_compression_algorithm::MYSQL_ZLIB:
            mysql->options.client_flag |= CLIENT_COMPRESS;
            mysql->options.compress = true;
            break;
          case enum_compression_algorithm::MYSQL_ZSTD:
            mysql->options.client_flag |= CLIENT_ZSTD_COMPRESSION_ALGORITHM;
            mysql->options.compress = true;
            break;
          case enum_compression_algorithm::MYSQL_UNCOMPRESSED:
            mysql->options.extension->connection_compressed = false;
            break;
          case enum_compression_algorithm::MYSQL_INVALID:
            break;  // report error
        }
        it++;
        cnt++;
      }
      if (cnt)
        EXTENSION_SET_STRING(&mysql->options, compression_algorithm,
                             static_cast<const char *>(arg));
      mysql->options.extension->total_configured_compression_algorithms = cnt;

https://github.com/mysql/mysql-server/blob/mysql-8.4.7/sql-common/client.cc#L8772-L8807

ここで現れる MYSQL 構造体という非常にわかりにくい名前のものがあるが、一度そのデータ構造について触れておく。MYSQL 構造体の宣言は次のようになっていて、ざっとみるとある程度理解できると思うが MySQL において何らかの通信を行う時には大抵必要となる情報が保持される。user, passwd あたりは言わずもがな、馴染み深いところで行くと thread_id 等だろうか。この構造体の利用用途は多岐にわたるので詳細をここで説明はしないが MySQL の C API を利用する場合、特に何らかのコネクションを利用する場合は確実に必要とされる構造体である。この構造体は mysql_real_connect 関数により接続を確立するために利用され、 mysql_init 関数により初期化することができる。

typedef struct MYSQL {
  NET net;                     /* Communication parameters */
  unsigned char *connector_fd; /* ConnectorFd for SSL */
  char *host, *user, *passwd, *unix_socket, *server_version, *host_info;
  char *info, *db;
  struct CHARSET_INFO *charset;
  MYSQL_FIELD *fields;
  struct MEM_ROOT *field_alloc;
  uint64_t affected_rows;
  uint64_t insert_id;      /* id if insert on table with NEXTNR */
  uint64_t extra_info;     /* Not used */
  unsigned long thread_id; /* Id for connection in server */
  unsigned long packet_length;
  unsigned int port;
  unsigned long client_flag, server_capabilities;
  unsigned int protocol_version;
  unsigned int field_count;
  unsigned int server_status;
  unsigned int server_language;
  unsigned int warning_count;
  struct st_mysql_options options;
  enum mysql_status status;
  enum enum_resultset_metadata resultset_metadata;
  bool free_me;   /* If free in mysql_close */
  bool reconnect; /* set to 1 if automatic reconnect */

  /* session-wide random string */
  char scramble[SCRAMBLE_LENGTH + 1];

  LIST *stmts; /* list of all statements */
  const struct MYSQL_METHODS *methods;
  void *thd;
  /*
    Points to boolean flag in MYSQL_RES  or MYSQL_STMT. We set this flag
    from mysql_stmt_close if close had to cancel result set of this object.
  */
  bool *unbuffered_fetch_owner;
  void *extension;
} MYSQL;

https://github.com/mysql/mysql-server/blob/mysql-8.4.7/include/mysql.h#L300-L338

というわけで話は前後してしまったが mysql_init 関数により MYSQL 構造体を初期化したのちに mysql_real_connect 関数を呼ぶ前、つまり接続を実際に確立する前に mysql_options 関数を使用して設定を与えることにより C API で提供される様々な接続制御を行うことができる。
前述の mysql_options 関数で行っていたことをこれらの情報を踏まえて整理すると、MySQL に対して接続を確立する際に何らかの接続圧縮制御を利用するという情報を設定したということである。

これで接続圧縮が利用できるようになったわけだがそれを実現しているのはさらに難解な仕組みが待っている。これは先に実際にパケットを圧縮解凍する処理を見つけたのでわかったことだが mysql_real_connect を呼び出した場合の処理で mysql_compress_context なるデータ構造を生成している。これはみるとある程度見えてくるが、zstd, zlib それぞれで圧縮する場合どのような圧縮が必要かという情報がまとまっている。例えば zstd の圧縮レベルなどもこのメンバにある mysql_zstd_compress_context で管理されている。

typedef struct mysql_compress_context {
  enum enum_compression_algorithm algorithm;  ///< Compression algorithm name.
  union {
    mysql_zlib_compress_context zlib_ctx;  ///< Context information of zlib.
    mysql_zstd_compress_context zstd_ctx;  ///< Context information of zstd.
  } u;
} mysql_compress_context;

https://github.com/mysql/mysql-server/blob/mysql-8.4.7/include/my_compress.h#L74-L80

参考までに mysql_compress_context_init がコールされる時点のバックトレースを記載する。

(lldb) bt
* thread #1, queue = 'com.apple.main-thread', stop reason = breakpoint 1.1
  * frame #0: 0x0000000100110268 mysql`mysql_compress_context_init(cmp_ctx=0x0000000100d95658, algorithm=MYSQL_ZSTD, compression_level=3) at my_compress.cc:62:24
    frame #1: 0x0000000100044220 mysql`csm_prep_select_database(ctx=0x000000016fdfdaa8) at client.cc:7128:5
    frame #2: 0x0000000100046c20 mysql`connect_helper(ctx=0x000000016fdfdaa8) at client.cc:6210:14
    frame #3: 0x000000010004e488 mysql`cli_connect(ctx=0x000000016fdfdaa8) at client.cc:6235:10
    frame #4: 0x0000000100047348 mysql`mysql_real_connect(mysql=0x0000000100905288, host=0x0000000000000000, user="root", passwd=0x0000000000000000, db=0x0000000000000000, port=0, unix_socket=0x0000000000000000, client_flag=66560) at client.cc:6270:10
    frame #5: 0x00000001000136dc mysql`sql_real_connect(host=0x0000000000000000, database=0x0000000000000000, user="root", (null)=0x0000000000000000, silent=0) at mysql.cc:4983:11
    frame #6: 0x0000000100005048 mysql`sql_connect(host=0x0000000000000000, database=0x0000000000000000, user="root", silent=0) at mysql.cc:5215:13
    frame #7: 0x0000000100003334 mysql`main(argc=0, argv=0x0000000100d8a078) at mysql.cc:1443:7
    frame #8: 0x0000000182a41d54 dyld`start + 7184

なぜこのようなことになっているのかというと、実際に接続を確立してデータをやり取りするのは MYSQL 構造体の中に含まれる NET 構造体を利用するためである。ここにはパケットのバッファやどこまで読んだ書いたのかというポジションからファイルディスクリプタと MySQL とのやりとりにおける一段低レイヤーな情報が集まっている。MySQL における送受信においてはこれを使っているのである。

typedef struct NET {
  MYSQL_VIO vio;
  unsigned char *buff, *buff_end, *write_pos, *read_pos;
  my_socket fd; /* For Perl DBI/dbd */
  /**
    Set if we are doing several queries in one
    command ( as in LOAD TABLE ... FROM MASTER ),
    and do not want to confuse the client with OK at the wrong time
  */
  unsigned long remain_in_buf, length, buf_length, where_b;
  unsigned long max_packet, max_packet_size;
  unsigned int pkt_nr, compress_pkt_nr;
  unsigned int write_timeout, read_timeout, retry_count;
  int fcntl;
  unsigned int *return_status;
  unsigned char reading_or_writing;
  unsigned char save_char;
  bool compress;
  unsigned int last_errno;
  unsigned char error;
  /** Client library error message buffer. Actually belongs to struct MYSQL. */
  char last_error[MYSQL_ERRMSG_SIZE];
  /** Client library sqlstate buffer. Set along with the error message. */
  char sqlstate[SQLSTATE_LENGTH + 1];
  /**
    Extension pointer, for the caller private use.
    Any program linking with the networking library can use this pointer,
    which is handy when private connection specific data needs to be
    maintained.
    The mysqld server process uses this pointer internally,
    to maintain the server internal instrumentation for the connection.
  */
  void *extension;
} NET;

https://github.com/mysql/mysql-server/blob/mysql-8.4.7/include/mysql_com.h#L915-L948

そしてここで最も厄介なものが末尾のメンバーとして宣言されている void *extension である。コメントを雑に理解するならば、ほぼどのような用途にでも使われると書かれているのである。これは大変な魔境となっている。
しかし残念ながら、前述の mysql_compress_context の実体はここにあるのである。接続圧縮に関する情報はこの extension なるメンバにあるというのが共通理解のようで、これをあろうことか NET_EXTENSION * にキャストする。そうすることで先ほど設定した compress_ctx を取得できるのである。

struct NET_EXTENSION {
  NET_ASYNC *net_async_context;
  mysql_compress_context compress_ctx;
};

https://github.com/mysql/mysql-server/blob/mysql-8.4.7/include/mysql_async.h#L199-L202

というわけでようやくこれで圧縮に辿り着けるのだが、ここまでの情報をクライアントが認識した上で実際にデータを MySQL に送信する際に送りたいパケットを圧縮するのである。
ここも全体像を先にある程度把握していないと理解が難しいので先に都合の良いところで止めたバックトレースを貼る。

(lldb) bt
* thread #1, queue = 'com.apple.main-thread', stop reason = breakpoint 2.1
  * frame #0: 0x000000010011052c mysql`my_compress(comp_ctx=0x00000001016dd658, packet="#", len=0x000000016fdfd8c0, complen=0x000000016fdfd820) at my_compress.cc:282:3
    frame #1: 0x0000000100063070 mysql`compress_packet(net=0x0000000100905288, packet="#", length=0x000000016fdfd8c0) at net_serv.cc:1270:7
    frame #2: 0x000000010006177c mysql`net_write_packet(net=0x0000000100905288, packet="#", length=39) at net_serv.cc:1319:19
    frame #3: 0x0000000100061640 mysql`net_flush(net=0x0000000100905288) at net_serv.cc:297:9
    frame #4: 0x0000000100062f80 mysql`net_write_command(net=0x0000000100905288, command='\x03', header="", head_len=2, packet="select @@version_comment limit 1", len=32) at net_serv.cc:915:55
    frame #5: 0x0000000100037e10 mysql`cli_advanced_command(mysql=0x0000000100905288, command=COM_QUERY, header="", header_length=2, arg="select @@version_comment limit 1", arg_length=32, skip_check=true, stmt=0x0000000000000000) at client.cc:1388:7
    frame #6: 0x00000001000490c4 mysql`mysql_send_query(mysql=0x0000000100905288, query="select @@version_comment limit 1", length=32) at client.cc:7948:13
    frame #7: 0x0000000100049d7c mysql`mysql_real_query(mysql=0x0000000100905288, query="select @@version_comment limit 1", length=32) at client.cc:8061:7
    frame #8: 0x0000000100022cc8 mysql`mysql_query(mysql=0x0000000100905288, query="select @@version_comment limit 1") at libmysql.cc:677:10
    frame #9: 0x000000010000567c mysql`server_version_string(con=0x0000000100905288) at mysql.cc:5367:10
    frame #10: 0x0000000100003454 mysql`main(argc=0, argv=0x00000001016d2078) at mysql.cc:1470:12
    frame #11: 0x0000000182a41d54 dyld`start + 7184

これのわかりやすいところから見ていくとまずここである。圧縮の true or false が true ならば compress_packet というまさにそれだろうという処理を呼び出す。

  const bool do_compress = net->compress;
  if (do_compress) {
    if ((packet = compress_packet(net, packet, &length)) == nullptr) {
      net->error = NET_ERROR_SOCKET_UNUSABLE;
      net->last_errno = ER_OUT_OF_RESOURCES;
      /* In the server, allocation failure raises a error. */
      net->reading_or_writing = 0;

https://github.com/mysql/mysql-server/blob/mysql-8.4.7/sql-common/net_serv.cc#L1317-L1323

compress_packet で重要な部分はここである。ここまでに紹介した compress_context なるものも登場し、my_compress というどう見ても低レイヤーな処理を次に呼び出す。そこをさらに辿っていくと送信したいパケットのポインタがありそれを zstd or zlib で圧縮しているのである。

static uchar *compress_packet(NET *net, const uchar *packet, size_t *length) {
  uchar *compr_packet;
  size_t compr_length = 0;
  const uint header_length = NET_HEADER_SIZE + COMP_HEADER_SIZE;

  compr_packet = (uchar *)my_malloc(key_memory_NET_compress_packet,
                                    *length + header_length, MYF(MY_WME));

  if (compr_packet == nullptr) return nullptr;

  memcpy(compr_packet + header_length, packet, *length);

  mysql_compress_context *compress_ctx = compress_context(net);

  /* Compress the encapsulated packet. */
  if (my_compress(compress_ctx, compr_packet + header_length, length,
                  &compr_length)) {
    /*
      If the length of the compressed packet is larger than the
      original packet, the original packet is sent uncompressed.
    */
    compr_length = 0;
  }

https://github.com/mysql/mysql-server/blob/mysql-8.4.7/sql-common/net_serv.cc#L1255-L1277

bool my_compress(mysql_compress_context *comp_ctx, uchar *packet, size_t *len,
                 size_t *complen) {
  DBUG_ENTER("my_compress");
  if (*len < MIN_COMPRESS_LENGTH) {
    *complen = 0;
    DBUG_PRINT("note", ("Packet too short: Not compressed"));
  } else {
    uchar *compbuf = my_compress_alloc(comp_ctx, packet, len, complen);
    if (!compbuf) DBUG_RETURN(*complen ? 0 : 1);
    memcpy(packet, compbuf, *len);
    my_free(compbuf);
  }
  DBUG_RETURN(0);
}

https://github.com/mysql/mysql-server/blob/mysql-8.4.7/mysys/my_compress.cc#L280-L293

これにて圧縮編は終了となる。MySQL の C API はこのようにしてネットワーク通じてをやり取りするパケットを圧縮・転送し、受け取るとそれを展開している。
解凍に関しては概ねこれの逆のことが行われていると考えて問題ない。むしろここまで共通化されているので今回紹介したものと別のパスが存在する方が不思議なぐらいだと思う。

またこのようなナイスな仕組みがあるので帯域をクソデカデータで食い潰してしまいそうな時は雑に接続圧縮制御を試してみると良いかもしれない。

余談

なぜ今回このような話をしたのかというと、これは Clone プラグインでも使われている*1ためである。
面白いことに Clone プラグインがリモートクローンを行う時は今回ここで紹介した方法をそのまま通る。つまり C API を利用することができて mysql_options 関数で圧縮アルゴリズムを指定し圧縮することができ、それをそのまま別のインスタンスで受け取り展開するということが本当にそのままできるのである。

これは諸々の事情によりレガシーな接続圧縮制御しか使えない、つまり zlib 圧縮しか Clone プラグインで zstd も扱えるようにナウい方の接続圧縮を使えるようにしようぜというパッチを送ったのでより理解が深まった。
bugs.mysql.com

おわりに

というわけで今回はいつになく高レイヤー(?)な話題でした。MySQL を取り巻くコネクションの実装なんかは初めて読むことになったのでなかなか難解でしたがこれもこれで面白いネタでした。
頼むからパッチ取り込まれてくれぇ。

*1:正確には前述のレガシーな接続圧縮制御が使われている