第2回 Non-blocking ソケットで IPv4/IPv6両対応 (クライアント側) [C++] [GTK4]

[2022.11] ページが長くなりすぎ。割った。

Note.

(2006.08.05 追記) (2022.04 追記)

下記のコードでは connect(), read() をnon-blockingモードにしている。しかし、getaddrinfo() も, ホスト名の解決を行うために時間が掛かる (ブロックする)。この考慮が足りない。

glibc 拡張の getaddrinfo_a(), Qt の QDnsLookup クラスは非同期だが、いずれも内部でスレッドを生成している。内部で getaddrinfo() を使う限り、このような実装しかない。libresolv の res_nsearch() (res_search() は非推奨.) も応答を待つ。

Ruby では、自前のリゾルバ Resolv::DNS クラスを作っている。すごい。ここまでやれば DNS lookupを非同期にできる。できるが、やりすぎ。

ということで、非同期 asynchronous I/O にするには, 自分でスレッドを生成して、getaddrinfo(), ブロッキングモードの connect() を呼び出してサーバに接続するようにすべき。Windowsでも同様の方法が推奨されている。(WSAAsyncGetHostByName の頁を参照。) もっとシンプルにできるだろう。

別ページで、Qt クライアントサンプルを解説する。ソケット通信はブロッキングモードにする。

余談。Qt はマルチスレッドでのワーカースレッド・メインスレッド間のメッセージのやり取りがシグナル/スロットを使うだけでよく、非常に簡単なので、わざわざ作るのが難しい non-blocking モードのソケットにするメリットもない。

他方、GTKは、Qtを真似したシグナルがスレッドを跨げないので、難儀する。gtkmm は Glib::Dispatcher クラスで, わざわざパイプを生成してスレッド間通信している。意外な機能で g_idle_add(), g_idle_add_full() が、名前と全然関係なく、メインスレッドでコールバックを呼び出してくれるので、これを使う。GTK4 になって非互換ばかりで, ますます無茶苦茶になっている。生産性も低く、もう実用にならない。

Windows では GetAddrInfoExW() (Unicode版のみ!) が非同期に対応している。が、サンプルが全然見当たらない。

(以下、2000.10.7, 2000.10.8の日記を加筆。)

ソケット関係の関数は、connect(), read(), サーバ側の accept() と、どれもこれもブロックする。ネットワーク越しのデータのやり取りは本質的に非同期なので、平行して入力がある場合、いちいちブロックするようにプログラムを書いてはいけない。

例えば、GUIプログラムの場合は、ソケットをブロックしていたら、マウスのクリック、画面の再描画などを受け付けられず、フリーズしてしまう。

ではどうするか。

  1. マルチスレッドにして, ソケットは blocking モードで接続, 通信する. 今どきはこれが簡単.
  2. connect() だけ non-blocking モードにして, 接続する。接続できたら blocking モードにして, select() または poll() で待ち合わせる.
  3. ずっと non-blocking モードで接続, 通信する.

ここでは、フレームワークが用意したコールバックの仕組みを使ってみる。どのようなフレームワークでも、ソケットに限らずさまざまな入力を捌く機構が用意されているので、それを使う。そうすれば、ソケットの入力を待ちながら画面を更新したりできるだろう。

Fedora Core 5 と gtk+ 2.8.20 という環境で試してみた。全体的にやっつけで作っているので、実用にしようと思うと、まだ考えるべきところが多いので注意。

[2022.11] GTK4 に更新。修正を要する非互換が多く, ヒドい.

Non-blocking ソケット

UNIXでは、ソケットに限らず、ファイルIOをブロックしないように設定できる。open() または fcntl() で O_NONBLOCK フラグを設定すると、non-blocking モードになる。ソケットでは, 後者の fcntl() 関数の F_SETFL コマンドで O_NONBLOCK フラグを設定できる。

Non-blockingモードでは、read(2), write(), send(), recv() 関数は、ブロックせずに、すぐに返ってくるようになる。戻り値とエラーコードで状況を調べる必要がある。EAGAIN または EWOULDBLOCK エラーが発生する。

connect() 関数もすぐに返ってくる。EINPROGRESS エラーになるかを確認する。

accept(2) も同様。listenキューが空でも、すぐに返ってる。EAGAIN または EWOULDBLOCK エラーが発生する。

EAGAIN と EWOULDBLOCK エラーは, 異なる値かもしれず、かつどちらのエラーもありうる。Linuxでは同じ値になっているが、移植性のため、両方を確認すること。

何を作るか

ごく簡単なGTK4 クライアントを作る。サーバから fortune テキストを得て、表示するだけ。

GTK アプリケーションの部分は、本題ではないので、割愛する。

接続要求関数

では、接続要求を投げるところから順に作っていこう.

Non-blocking モードの設定

ソケットを non-blocking モードにするには fcntl() を使う。F_GETFL コマンドで現在のフラグを読み出し、F_SETFL で設定する。

C++
[RAW]
  1. /**
  2. * @param mode 非0 = into non-blocking mode.
  3. * @return If failed, -1.
  4. */
  5. int non_blocking( SOCKET sock, int mode )
  6. {
  7. assert( sock != INVALID_SOCKET );
  8. #ifndef _WIN32
  9. // UNIX
  10. int oldflags = fcntl(sock, F_GETFL, 0 /* dummy */);
  11. return fcntl(sock, F_SETFL,
  12. mode ? (oldflags | O_NONBLOCK) : (oldflags & ~O_NONBLOCK) );
  13. #else
  14. // Windows では「現在のモード」を得る方法はない
  15. u_long iMode = mode ? 1 : 0;
  16. int iResult = ioctlsocket( sock, FIONBIO, &iMode);
  17. if ( iResult != NO_ERROR )
  18. return -1;
  19. return 0;
  20. #endif
  21. }

接続要求

getaddrinfo() が複数のソケットアドレスを返す. 通常は IPv6 アドレスと IPv4 アドレスの二つ。両方に対して接続要求を出すので、複雑になる.

Non-blocking ソケットでは、connect() がすぐに返ってくる。サーバがIPv4対応かIPv6対応か分からない状況では、ソケットを2本使う必要がある。両方でconnectを試みて、接続できたほうを生かす。

C++
[RAW]
  1. // <var>hostname</var> からIPアドレスを引き, 全部に接続要求する. Non-blocking 版.
  2. // 実際に接続したときに on_connect() が呼び出されるようにする.
  3. // @return 少なくとも一つが接続要求が成功 = true
  4. bool start_connect_to_server_nonblock(const char* hostname, int port)
  5. {
  6. assert( hostname );
  7. if (port <= 0 || port > 65535 )
  8. return false;
  10. struct addrinfo hints;
  11. memset(&hints, 0, sizeof(hints));
  12. hints.ai_family = AF_UNSPEC; // IPv4/IPv6 両対応
  13. hints.ai_socktype = SOCK_STREAM;
  14. hints.ai_flags = AI_NUMERICSERV;
  16. char service[11];
  17. sprintf(service, "%d", port);
  19. struct addrinfo* res = NULL;
  20. // ●ここで待ちが入る。アカン!!
  21. int err = getaddrinfo(hostname, service, &hints, &res);
  22. if (err != 0) {
  23. fprintf(stderr, "getaddrinfo() failed: %s\n", gai_strerror(err));
  24. error_dialog("ホストまたはポートが不正です。");
  25. return false;
  26. }

冒頭 Note. のとおり, getaddrinfo() がブロックしてしまうため、このサンプルは傷がある。無理やりいくなら、ここだけマルチスレッドにする。

connect() はブロックされず、EINPROGRESS エラーを発生させる。それを確認して、メインループでイベントを待ち合わせる.

接続に成功すると書き込み可能状態 POLLOUT になるので，そのようにコールバック関数を登録する。start_watch() 関数でおこなっている。

C++
[RAW]
  1. SOCKET sockfd = INVALID_SOCKET;
  2. bool succeeded = false;
  3. for (addrinfo* ai = res; ai; ai = ai->ai_next) {
  4. sockaddr_print("connect...", ai->ai_addr, ai->ai_addrlen);
  6. sockfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
  7. if ( sockfd == INVALID_SOCKET ) {
  8. freeaddrinfo(res);
  9. return false;
  10. }
  12. non_blocking(sockfd, 1); // 非ブロックモードに切り替え.
  13. int r = ::connect(sockfd, ai->ai_addr, ai->ai_addrlen);
  14. if (r < 0) {
  15. if (errno == EINPROGRESS || errno == EINTR ) { // 即時確立しなかった
  16. // Non-blocking モードのときは, シグナル割り込みは EINPROGRESS
  17. // と同じでよいみたい.
  18. // 成功. 書き込み可能になるまで待つ.
  19. pair<GIOChannel*, guint> ev =
  20. start_watch(sockfd, G_IO_OUT, on_connect, NULL);
  21. wait_connect_events.insert(ev);
  22. // Non-blocking 版では, 全部を試す.
  23. succeeded = true;
  24. continue;
  25. }
  26. closesocket(sockfd);
  27. sockfd = INVALID_SOCKET;
  28. continue;
  29. }
  30. else {
  31. // 即時成功の場合, 他のを close して, g_connection を設定.
  32. on_connect(NULL, G_IO_OUT, NULL);
  33. return true;
  34. }
  35. }
  36. freeaddrinfo(res); // もういらん.
  38. return succeeded;
  39. }

GTK でコールバック関数を登録する start_watch() 関数を見ておこう。

C++
[RAW]
  1. pair<GIOChannel*, guint>
  2. start_watch(SOCKET sockfd, int condition, GIOFunc proc, gpointer user_data)
  3. {
  4. assert(sockfd != INVALID_SOCKET);
  6. // GIOChannel で wrap する
  7. GIOChannel* channel = g_io_channel_unix_new(sockfd);
  8. guint io_watch = g_io_add_watch_full(channel, G_PRIORITY_DEFAULT,
  9. (GIOCondition) condition,
  10. proc,
  11. user_data,
  12. on_io_destroy);
  13. assert(io_watch);
  14. // g_io_add_watch() 内でもrefされるので、ここでは解放する
  15. g_io_channel_unref(channel);
  17. return make_pair(channel, io_watch);
  18. }

gdk_input_add() 関数が廃止され, g_io_add_watch_full() を使う. 代替関数なしにドンドコ deprecated にしていくのは止めてほしい。ファイルディスクリプタをそのまま登録するのではなく, GIOChannel で wrap する.

余談. g_io_channel_unix_new() の引数の型が int 決め打ちで、64bit Windows では動かない. 何のために wrap したのか意味不明. GPollFD 型は 64bit Windows では64bit ハンドルになっているので、修正漏れか.

接続成功 → イベント登録

接続に成功したら，読み込み可能状態でコールバックしてもらうように登録し直す。あとはコールバックされたら read() すればよい。

C++
[RAW]
  1. // Callback. 接続した or エラー. 成功の場合, g_connection を設定する.
  2. // @param source NULL の場合がある.
  3. gboolean on_connect(GIOChannel* source, GIOCondition condition, gpointer data)
  4. {
  5. print_condition(__func__, condition); // DEBUG
  7. if ( condition & (G_IO_ERR | G_IO_HUP) ) {
  8. // 自分だけ削除する.
  9. SOCKET fd = g_io_channel_unix_get_fd(source);
  10. closesocket(fd);
  12. wait_connect_events.erase(source);
  13. if (wait_connect_events.size() == 0) {
  14. error_dialog("接続に失敗しました。");
  15. g_idle_add(on_connect_error, (void*) -1); // emit.
  16. }
  17. return FALSE; // FALSE = The event source to be removed.
  18. }
  20. assert( wait_connect_events.size() > 0 );
  22. // ほかのを全部 close する
  23. for (auto i = wait_connect_events.begin();
  24. i != wait_connect_events.end(); i++) {
  25. if (i->first == source)
  26. continue;
  27. SOCKET fd = g_io_channel_unix_get_fd(i->first);
  28. closesocket(fd);
  29. g_source_remove(i->second); // 内部で g_io_channel_unref() される.
  30. }
  31. wait_connect_events.clear();
  33. // Callback を改めて登録する.
  34. SOCKET sockfd = g_io_channel_unix_get_fd(source);
  35. g_connection = start_watch(sockfd, G_IO_IN, on_ready_to_read, NULL);
  37. return FALSE; // FALSE = The event source to be removed.
  38. }

複数の接続要求を同時に走らせているので、接続できた場合はそれ以外の全部、接続できなかった場合はそのソケットを閉じていく。不必要に複雑になっている。

読み取って表示

EOF の場合は, エラーではなく, 読み取り可能かつ recv() の戻り値が 0 になる。

C++
[RAW]
  1. // Callback. GIOChannel のほうが引数になる. -> 直接 g_source_remove() できない.
  2. gboolean on_ready_to_read(GIOChannel* source, GIOCondition condition,
  3. gpointer data)
  4. {
  5. print_condition(__func__, condition); // DEBUG
  7. if ( condition & (G_IO_ERR | G_IO_HUP) ) {
  8. error_dialog("読み込み中にエラー発生");
  9. close_and_notify(false);
  10. return FALSE; // FALSE = The event source to be removed.
  11. }
  13. if (condition & G_IO_IN) {
  14. SOCKET fd = g_io_channel_unix_get_fd(g_connection.first);
  15. char buf[1000];
  16. int bytes_read = recv(fd, buf, sizeof(buf) - 1, 0);
  17. if (bytes_read < 0) {
  18. perror("recv() failed");
  19. error_dialog("recv() でエラー発生");
  20. close_and_notify(false);
  21. return FALSE; // FALSE = The event source to be removed.
  22. }
  23. else if ( bytes_read == 0 ) {
  24. // EOF
  25. close_and_notify(true);
  26. return FALSE;
  27. }
  29. assert( bytes_read > 0 );
  30. buf[bytes_read] = '\0';
  31. GtkTextBuffer* buffer = gtk_text_view_get_buffer(top_window->textView);
  32. assert(buffer);
  33. gtk_text_buffer_insert_at_cursor(buffer, buf, bytes_read);
  34. }
  36. return TRUE; // 再利用する.
  37. }

read() は次のいずれかで返る。

  1. 引数として与えたバッファが埋まった
  2. 書き込み側のwrite()の切れ目
  3. 書き込み側でソケットが閉じられた

上記の 2. から, バッファのバイト長より読めた長さが短くても、ソケットが閉じられたとは限らない。同様にread()の戻り値がバッファのバイト長と同じでも，まだデータがあるとは限らない。

ブロッキングモードの場合, read() の戻り値がたまたまバッファのバイト長と同じで、まだデータがあると思ってさらに read() するとブロックしてしまう。

一回だけ read() したら読み込みのルーチンを終了し，メインループに戻すようにする。メインループの待ち合わせでデータがあるかどうかを検査し、また読み込みルーチンを呼び出してもらう。

開始ボタン

あとは、開始ボタンのハンドラ。マルチスレッド版のコードと共用。

callbacks.cpp

C++
[RAW]
  1. // Callback
  2. void on_get_button_clicked(GtkButton* button, gpointer user_data)
  3. {
  4. #ifdef USE_THREAD
  5. if (worker_thread)
  6. return;
  7. #endif
  9. // GTK4: gtk_entry_get_text() がいきなり廃止。
  10. GtkEntryBuffer* b = gtk_entry_get_buffer( GTK_ENTRY(top_window->hostEntry) );
  11. worker_param.hostname = gtk_entry_buffer_get_text(b);
  12. b = gtk_entry_get_buffer( GTK_ENTRY(top_window->portEntry) );
  13. const char* port_str = gtk_entry_buffer_get_text(b);
  14. worker_param.port = atoi(port_str);
  15. if ( worker_param.port <= 0 || worker_param.port > 65535 ) {
  16. GtkDialog* dialog = GTK_DIALOG( gtk_message_dialog_new(top_window->self,
  17. GTK_DIALOG_DESTROY_WITH_PARENT,
  18. GTK_MESSAGE_ERROR,
  19. GTK_BUTTONS_CLOSE,
  20. "Port error") );
  21. g_signal_connect(dialog, "response",
  22. G_CALLBACK(gtk_window_destroy), NULL);
  24. //GTK4: gtk_dialog_run() は廃止.
  25. gtk_widget_show( GTK_WIDGET(dialog) );
  26. return;
  27. }
  29. #ifdef USE_THREAD
  30. // g_thread_new() は使うな. glib 2.32以降.
  31. // g_thread_create() は廃れた.
  32. GError* error = nullptr;
  33. worker_thread = g_thread_try_new("worker", start_get_fortune, &worker_param,
  34. &error);
  35. if ( !worker_thread ) {
  36. g_printerr("Error creating thread: %s\n", error->message);
  37. g_clear_error(&error);
  38. return;
  39. }
  40. #else
  41. if (!start_connect_to_server_nonblock(
  42. worker_param.hostname.c_str(), worker_param.port)) {
  43. return;
  44. }
  45. #endif
  47. gtk_widget_set_sensitive( GTK_WIDGET(top_window->hostEntry), FALSE );
  48. gtk_widget_set_sensitive( GTK_WIDGET(top_window->portEntry), FALSE );
  49. gtk_widget_set_sensitive( GTK_WIDGET(button), FALSE );
  50. }

関数をいきなり廃止したりするの、本当に止めてほしい。