第4回 libuv C10K〜C100K サーバサンプル [C++]

C10K問題 (C10K problem) を乗り越えるため, とりあえず 1,000 同時接続を超えるサンプル.

同時接続数とパフォーマンス向上の技術

POSIX API

select(2) は上限が低い. 待ち合わせする file descriptor の「値」の上限が FD_SETSIZE マクロで決まる。

OS によっては、アプリケィションのコンパイル時に, FD_SETSIZE マクロ値をコンパイラオプションで与えて変更することで、上限を引き上げられる。FreeBSD はそのようになっている。

しかし、Linux は 1024 で完全に決め打ちになっている。<sys/select.h> (glibc 提供ファイル) から参照される <linux/posix_types.h> (Linux kernel のファイル) で, __FD_SETSIZE も決め打ちになっている。

File descriptor はソケットだけでなく, 標準入力 = 0, ... のようにファイルを開いても消費される。そのため、せいぜい 1,000 程度しか待ち受けできない。同時接続数が不明な場合は select() を使ってはならない。

POSIX 標準では poll(2) が 1,000 接続の上限を超えて待ち受けできる。しかし、インタフェイスの設計がよくなく、パフォーマンスは select() 程度。

各OSの取組み

Linux での非同期I/O手法とそれらのパフォーマンス: Asynchronous C++ History of Time マルチスレッドでも 10,000 接続ぐらいまではイケる。Linux epoll はよい。

それぞれの OS がバラバラな方法で、多くの接続を「効率的に」待ち受けできるように新しいAPIを導入した。ソケットはカーネルで実装されるので、通常、システムコールとして提供される.

  • UNIX
    • Linux 2.6+ epoll(7) システムコール. epoll_create(2)
    • Solaris7 11/99+ poll(7d). /dev/poll デヴァイス. HP-UX でも実装されている. dvpoll 構造体.
    • Solaris10 Event Ports イベント. port_create(). 問題あり, /dev/poll を使うべきか?
    • FreeBSD 4.1以降 kqueue(2): kqueue(), kevent() システムコール. NetBSD 1.6.1 (2003年), NetBSD 2.0 (2004年) 以降, OpenBSD 2.9以降でも実装.
    • AIX pollset API. pollset_create() など.
  • Windows I/O Completion Port (IOCP). コピー回数が少ない. スレッドを効率的に動かす.

ポータブルなライブラリ

てんでバラバラなので、アプリケィション開発には, これらをカバーするライブラリを使おう。動かしたい環境 (の幅) とオーバヘッドの少なさでどれを使うべきか決まる。

  1. libev - a full-featured and high-performance event loop. UNIX 限定であればこれ. パフォーマンスも各OSのシステムコールを直接叩くのと遜色ない.
  2. libuv - a cross-platform asynchronous I/O. Windows でも動かしたい場合はこれ。パフォーマンスは libev より劣るが libevent よりはよい.
  3. POCO C++ Libraries - 組込みOS (embedded OS) でも動く. 軽いライブラリが必要な場合はこれ。パフォーマンスは libevent = POCO = asio で, libev, libuv より劣る.
  4. ▲ libevent - 古い. 新しいプロジェクトでは libev を使え.
  5. ▲ (non-Boost) Asio and Boost.Asio.

このほか, libhv というのもあって, Windows IOCP もサポートしつつ、libev と同等のパフォーマンスと主張している: GitHub - ithewei/libhv: 🔥 比libevent、libuv更易用的网络库。A c/c++ network library for developing TCP/UDP/SSL/HTTP/WebSocket/MQTT client/server. Fedora 37, Debian 11 にはパッケージがない。試していない。

Current or soft limit

以下で作るサンプルは、そのまま実行してもエラーになる。Soft limits がログイン端末ごとに設定されている。ulimit コマンドで表示できる (open files が 1,024 になっている)。標準入出力なども file descriptor を消費するので、これで 1,000 を少し超えた辺りでエラーになる。 

$ ulimit -a
real-time non-blocking time  (microseconds, -R) unlimited
core file size              (blocks, -c) unlimited
data seg size               (kbytes, -d) unlimited
scheduling priority                 (-e) 0
file size                   (blocks, -f) unlimited
pending signals                     (-i) 23404
max locked memory           (kbytes, -l) 8192
max memory size             (kbytes, -m) unlimited
open files                          (-n) 1024
pipe size                (512 bytes, -p) 8
POSIX message queues         (bytes, -q) 819200
real-time priority                  (-r) 0
stack size                  (kbytes, -s) 8192
cpu time                   (seconds, -t) unlimited
max user processes                  (-u) 23404
virtual memory              (kbytes, -v) unlimited
file locks                          (-x) unlimited

サーバ側の実行前に $ ulimit -S -n 110000 などしておく.

同時接続数 10,000 (C10K) 〜 C100K サーバ

libuv で C10K サーバを作ってみる。たかだか一つだけのクライアントから接続を受け付けるサンプルは Web 上にいくらでも見つかるが、複数接続で C++ のサンプルは、意外と見当たらない。

Echo server を作る。

main() から見ていこう.

server-common.cpp

C++
[RAW]
  1. constexpr int PORT = 7000;
  2. constexpr int BACKLOG = 100;
  4. // イベントループ. グローバル.
  5. uv_loop_t* g_loop = NULL;
  7. // サーバ. グローバル.
  8. uv_tcp_t* g_server = NULL;
  10. // これをサンプルで切り替える.
  11. extern void on_connection( uv_stream_t* server, int status );
  13. int main()
  14. {
  15. struct rlimit rlim;
  16. getrlimit(RLIMIT_NOFILE, &rlim);
  17. printf("soft = %ld, hard = %ld, FD_SETSIZE = %d\n",
  18. rlim.rlim_cur, rlim.rlim_max, FD_SETSIZE);
  20. // イベントループ
  21. g_loop = uv_default_loop();
  23. g_server = (uv_tcp_t*) malloc(sizeof(uv_tcp_t));
  24. uv_tcp_init( g_loop, g_server ); // まだソケットは作られない.
  26. printf("Waiting port = %d\n", PORT);
  27. int r = tcp_setup_server(g_server, NULL, PORT);
  28. if (r) {
  29. fprintf(stderr, "tcp_setup_server() failed.\n");
  30. return 1;
  31. }
  33. r = uv_listen( (uv_stream_t*) g_server, BACKLOG, on_connection );
  34. if (r) {
  35. // Error
  36. fprintf( stderr, "uv_listen() failed: %s\n", uv_strerror(r) );
  37. return 1;
  38. }
  40. // ループ開始
  41. r = uv_run(g_loop, UV_RUN_DEFAULT);
  42. printf("Exit uv_run()\n"); // uv_close(server) でループを抜ける.
  44. return r;
  45. }

uv_default_loop() でイベントループを得る.

ソケットを wrap する uv_tcp_t 変数をイベントループに結びつける。libuv はイベントループが中心になる。ソケットを tcp_setup_server() 内で bind し, listen して接続を受け付けたら on_connection() をコールバックさせる.

uv_run() でループ開始する。

次に, tcp_setup_server() 関数. ソケットを開いて bind する. uv_tcp_t のメソッドである uv_tcp_bind() は, BSDソケット関数の socket(), bind() を一体にしたもの。

server-common.cpp

C++
[RAW]
  1. #include <stdio.h>
  2. #include <assert.h>
  3. #include <string.h>
  4. #include <uv.h>
  5. #include <sys/resource.h> // getrlimit()
  6. #include <stdlib.h>
  8. /**
  9. * TCP で bind() までを行う. IPv6対応.
  10. * uv_stream_t 型を基底クラスとして, uv_tcp_t, uv_pipe_t および uv_tty_t が派生
  11. * クラス.
  12. * @param node bind() するホスト名. NULL の場合, INADDR_ANY, IN6ADDR_ANY_INIT.
  13. * @param port ポート番号. 0 の場合, 空いているポートを割り当てる.
  14. *
  15. * @return 成功 = 0, エラーの場合 = -1
  16. */
  17. int tcp_setup_server(uv_tcp_t* server, const char* node, int port)
  18. {
  19. assert(server);
  20. if (port < 0)
  21. return -1;
  23. // uv_ip4_addr() を使うサンプルは古い!!
  24. struct addrinfo hints;
  25. struct addrinfo* res = NULL;
  27. memset(&hints, 0, sizeof hints);
  28. if (!node)
  29. hints.ai_family = AF_INET6; // IPv4/IPv6両対応
  30. hints.ai_socktype = SOCK_STREAM;
  31. // AI_PASSIVE をセットして node = NULLのときは, INADDR_ANY, IN6ADDR_ANY_INIT.
  32. hints.ai_flags = AI_PASSIVE | AI_NUMERICSERV;
  34. // (node, service) の両方を NULL にしてはいけない.
  35. char service[11];
  36. sprintf(service, "%d", port);
  37. int err = getaddrinfo(node, service, &hints, &res);
  38. if (err != 0) {
  39. fprintf( stderr, "getaddrinfo() failed: %s\n", gai_strerror(err) );
  40. return -1;
  41. }
  43. // socket(), bind() を一体にしたもの.
  44. int r = uv_tcp_bind(server, res->ai_addr, 0);
  45. freeaddrinfo(res);
  46. if (r) {
  47. fprintf(stderr, "uv_tcp_bind() failed: %s\n", uv_strerror(r) );
  48. return -1;
  49. }
  51. return 0;
  52. }

コールバックされる on_connection() は次のようにする。

よくあるサンプルでは, uv_accept() で得たソケットをグローバル変数に格納したりする。それでは一つのクライアントからの接続にしか対応できない。ClientSession クラスを作って、配列に格納するようにする。

multi-tcp-server.cpp

C++
[RAW]
  1. map<uv_tcp_t*, class ClientSession*> ClientSession::_session_list;
  3. void on_connection( uv_stream_t* server, int status )
  4. {
  5. if ( status != 0 ) {
  6. // エラーの場合, エラー名を得る uv_err_name() or エラーメッセージを得る
  7. // uv_strerror() を使う.
  8. fprintf(stderr, "New connection error: %s\n", uv_strerror(status) );
  9. return;
  10. }
  12. uv_tcp_t* client = (uv_tcp_t*) malloc(sizeof(uv_tcp_t));
  13. uv_tcp_init( g_loop, client );
  15. int r = uv_accept(server, (uv_stream_t*) client);
  16. if (r) {
  17. fprintf(stderr, "uv_accept() failed: %s\n", uv_strerror(r) );
  18. free(client);
  19. return;
  20. }
  22. ClientSession* session = ClientSession::new_instance(client); // client所有渡す
  23. r = session->read_start();
  24. assert( r == 0 );
  25. }

BSD ソケットの accept() を wrap する uv_accept() で, 新しくクライアントとの接続用のソケットができる。

その新しいソケットで, 読み取り可能になったときに呼び出されるコールバック関数を登録、そのまま待ち受け開始する。ClientSession クラスの read_start() までを作る.

C++
[RAW]
  1. class ClientSession: private noncopyable
  2. {
  3. // uv_tcp_t < uv_stream_t < uv_handle_t
  4. uv_tcp_t* m_handle; // 所有する
  6. static map<uv_tcp_t*, class ClientSession*> _session_list;
  8. protected:
  9. ClientSession(uv_tcp_t* h): m_handle(h) {
  10. assert( h != NULL );
  11. }
  13. public:
  14. static ClientSession* new_instance(uv_tcp_t* h) {
  15. assert(h);
  16. ClientSession* self = new ClientSession(h);
  17. _session_list.insert(make_pair(h, self));
  18. return self;
  19. }
  21. ~ClientSession() {
  22. assert(m_handle);
  23. _session_list.erase(m_handle);
  24. free(m_handle); m_handle = NULL;
  25. }
  27. // @return If error, non-zero
  28. int read_start() {
  29. assert( m_handle != NULL );
  30. // Read data from an incoming stream. alloc_buffer() で領域を確保する.
  31. return uv_read_start( (uv_stream_t*) m_handle, _alloc_buffer,
  32. (uv_read_cb) _on_read );
  33. }

ハンドルに対してコールバック関数を uv_read_start() 関数で登録すればOK.

次のコールバック関数を書いていく。配列のどのセッションか特定する _on_read() と, 実際のコールバック関数の on_receive() を書く.

C++
[RAW]
  1. /**
  2. * Callback.
  3. * EOF の場合, uv__stream_eof() から呼び出される. 第2引数が UV_EOF.
  4. * この場合も, 領域確保済み.
  5. * それ以外, uv__read() から呼び出される.
  6. * @param nread < 0 の場合, EOF or エラー. > 0 の場合, 読み出せる.
  7. * == 0 の場合, ソケットが non-blocking で読み出せるデータなし.
  8. */
  9. static void _on_read( uv_stream_t* client, ssize_t nread, uv_buf_t* buf ) {
  10. ClientSession* v = _session_list.at((uv_tcp_t*) client);
  11. v->on_receive(nread, buf);
  13. // エラーの場合も, 領域を解放しなければならない. ●●重い処理.
  14. if (buf->base) {
  15. free(buf->base); buf->base = NULL;
  16. }
  17. }

libuv の特徴は、読み込み可能になったらコールバックされるのではなく、Windows 流に合わせて、読み込みが完了してからコールバックされる.

C++
[RAW]
  1. void on_receive(ssize_t nread, const uv_buf_t* buf) {
  2. assert(buf);
  4. if (nread < 0) {
  5. if (nread != UV_EOF) {
  6. fprintf(stderr, "recv() EOF or error: %s\n", uv_err_name(nread) );
  7. }
  8. // If error, *user* should call uv_close().
  9. shutdown();
  10. }
  11. else if (nread > 0) {
  12. if (buf->base[0] == 'X') {
  13. printf("Get client shutdown message.\n");
  14. shutdown();
  15. }
  16. else if (buf->base[0] == 'Z') {
  17. printf("Get server shutdown message.\n");
  18. server_shutdown();
  19. }
  20. else {
  21. // 書き込みもややこしい.
  22. uv_write_t* req = (uv_write_t*) malloc(sizeof(uv_write_t));
  23. uv_buf_t wbuf = uv_buf_init(buf->base, nread); // ローカル変数でよい.
  24. uv_write(req, (uv_stream_t*) m_handle, &wbuf, 1, _after_write);
  25. }
  26. }
  27. }

_alloc_buffer() で, 毎回, メモリを確保する。uv_read_start() でコールバックを登録していた。

C++
[RAW]
  1. /**
  2. * Callback. stream->alloc_cb(stream, 64 * 1024, &buf) の形で呼び出される.
  3. * @param handle stream
  4. * @param suggested_size 64 * 1024 固定.
  5. * @param [out] buf バッファを格納. .base == NULL || .len == 0 で戻すと,
  6. * read_cb() に UV_ENOBUFS エラーを通知.
  7. */
  8. static void _alloc_buffer( uv_handle_t* handle, size_t suggested_size,
  9. uv_buf_t* buf)
  10. {
  11. //printf("alloc_buffer() called: .\n");
  12. buf->base = (char*) malloc(suggested_size); // ●●重い! 自分で管理が必要.
  13. buf->len = suggested_size;
  14. }

1回の読み込みごとにメモリを解放するようになっており、非常に重い。完全に固定長のバッファなので、別にメモリプールを作るとパフォーマンスが上がる可能性が高い。

クライアント

次に、クライアント. こちらは libuv を使っていない。少ない台数でテストできるようにするため, 子プロセスをドンドン走らせていく。

Linux で試したところ、100 プロセスにすると、接続に失敗する子プロセスが発生する。別のパラメータの上限がありそう. -> Ephemeral port (エフェメラルポート) を使い果たしたエラー.

接続エラーになる少し前の状態。この後にポートを使い果たしたか。 

$ ss -tan | awk '{print $1}' | sort | uniq -c
 107693 ESTAB
      8 FIN-WAIT-2
      9 LISTEN
      1 State
    233 TIME-WAIT
C++
[RAW]
  1. constexpr int CLIENT_SIZE = 1000;
  2. // C100K => 50 は OK. 100 にすると失敗発生しだす.
  3. // connect() failed: Cannot assign requested address
  4. // Addr is...: AF_INET6 [::1]:7000
  5. constexpr int CHILD_NUMBER = 100;
  7. // @return 0 が成功.
  8. int make_children(const char* argv0, const char* hostname)
  9. {
  10. assert(hostname);
  12. #ifndef _WIN32
  13. struct rlimit rlim;
  14. getrlimit(RLIMIT_NOFILE, &rlim);
  15. printf("soft = %ld, hard = %ld, FD_SETSIZE = %d\n",
  16. rlim.rlim_cur, rlim.rlim_max, FD_SETSIZE);
  17. #endif
  19. for (int k = 0; k < CHILD_NUMBER; k++) {
  20. #ifndef _WIN32
  21. // UNIX: fork() + exec*() が伝統的な方法. posix_spawn(3) でもよい.
  22. pid_t pid = fork();
  23. if (pid == -1) {
  24. fprintf(stderr, "fork() failed.\n");
  25. return 1;
  26. }
  27. else if (pid == 0) {
  28. // child
  29. execl(argv0, argv0, "-c", hostname, NULL); // restart
  30. abort(); // ここには来ない
  31. }
  32. // 親プロセス
  33. #else
  34. // Windows: _spawn* 系かCreateProcess() を直接使う.
  35. _spawnl(_P_NOWAIT, argv0, argv0, "-c", hostname, NULL);
  36. #endif
  37. }
  38. return 0;
  39. }

子プロセスで、実際に通信する。接続を繋ぎっぱなしにするため、適宜、待ちを入れる.

C++
[RAW]
  1. SOCKET sockfd[CLIENT_SIZE];
  3. // @return 0 が成功.
  4. int test_connect_server(const char* hostname)
  5. {
  6. assert(hostname);
  8. #ifdef _WIN32
  9. // Initialize Winsock
  10. WSADATA wsaData;
  11. int iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);
  12. if (iResult != 0) {
  13. printf("WSAStartup failed: %d\n", iResult);
  14. return 1;
  15. }
  16. #endif
  18. for (int i = 0; i < CLIENT_SIZE; i++) {
  19. sockfd[i] = connect_to_server(hostname, 7000);
  20. if (sockfd[i] == INVALID_SOCKET) {
  21. fprintf(stderr, "%d: connect_to_server() failed.\n", i); //=> 1020 でコケる
  22. return 1;
  23. }
  24. }
  26. const char ch = 'A';
  27. for (int i = 0; i < CLIENT_SIZE; i++) {
  28. // 送受信
  29. send(sockfd[i], &ch, 1, 0);
  30. }
  31. printf("send finished.\n");
  32. #ifndef _WIN32
  33. sleep(5);
  34. #else
  35. Sleep(5 * 1000);
  36. #endif
  37. for (int i = 0; i < CLIENT_SIZE; i++) {
  38. char rch = '\0';
  39. recv(sockfd[i], &rch, 1, 0);
  40. if (ch != rch) {
  41. fprintf(stderr, "%d: char from server = %c\n", i, rch);
  42. exit(1);
  43. }
  44. }
  45. printf("receive finished.\n");
  46. #ifndef _WIN32
  47. sleep(5);
  48. #else
  49. Sleep(5 * 1000);
  50. #endif
  51. /*
  52. #if 0
  53. // 明示的に shutdown する
  54. ch = 'X';
  55. send(sockfd, &ch, 1, 0);
  56. #endif
  57. // 明示的に server を shutdown する
  58. ch = 'Z';
  59. send(sockfd, &ch, 1, 0);
  60. */
  61. for (int i = 0; i < CLIENT_SIZE; i++) {
  62. closesocket(sockfd[i]);
  63. }
  65. return 0;
  66. }