プロセス¶
libuvは、プラットフォーム間の違いを抽象化し、ストリームや名前付きパイプを使って子プロセスと通信できる、充実した子プロセス管理機能を提供します。
Unixでよく使われる考え方は、各プロセスが1つのことを行い、それをうまく行うというものです。その場合、プロセスは、タスクを達成するために複数の子プロセスを使うことがよくあります（シェルでパイプを使うのと似ています）。メッセージを使うマルチプロセスモデルは、スレッドと共有メモリを使うモデルよりも、考えやすい場合があります。
イベント駆動プログラムに対するよくある批判は、現代のコンピュータの複数コアを活用できないというものです。マルチスレッドプログラムでは、カーネルがスケジューリングを行い、異なるスレッドを別々のコアに割り当てて性能を高められます。しかし、イベントループにはスレッドが1つしかありません。代わりに複数のプロセスを起動し、各プロセスがイベントループを実行して、それぞれを別のCPUコアに割り当てるという方法があります。
子プロセスの起動¶
最も単純なのは、プロセスを起動して、終了したときを知りたい場合です。これには、uv_spawnを使います。
spawn/main.c
1uv_loop_t *loop;
2uv_process_t child_req;
3uv_process_options_t options;
4int main() {
5 loop = uv_default_loop();
6
7 char* args[3];
8 args[0] = "mkdir";
9 args[1] = "test-dir";
10 args[2] = NULL;
11
12 options.exit_cb = on_exit;
13 options.file = "mkdir";
14 options.args = args;
15
16 int r;
17 if ((r = uv_spawn(loop, &child_req, &options))) {
18 fprintf(stderr, "%s\n", uv_strerror(r));
19 return 1;
20 } else {
21 fprintf(stderr, "Launched process with ID %d\n", child_req.pid);
22 }
23
24 return uv_run(loop, UV_RUN_DEFAULT);
25}
注記
optionsはグローバル変数なので、暗黙にゼロ初期化されます。optionsをローカル変数に変更する場合は、未使用のすべてのフィールドが空になるよう、忘れずに初期化してください。
uv_process_options_t options = {0};
uv_process_t構造体はハンドルとしてのみ機能し、すべてのオプションは、uv_process_options_tで設定します。単にプロセスを起動するには、fileとargsフィールドだけを設定すれば十分です。fileは、実行するプログラムです。uv_spawnは内部で、execvp(3)を使うため、完全なパスを指定する必要はありません。最後に、基になる規約に従い、引数の配列は引数の数より1要素多くし、最後の要素をNULLにする必要があります。
uv_spawnの呼び出し後、uv_process_t.pidには、子プロセスのプロセスIDが格納されます。
終了コールバックには、終了ステータスと、終了の原因になったシグナルの種類が渡されます。
終了コールバックより前には、決して、uv_closeを呼び出さないことが重要です。
spawn/main.c
1
2void on_exit(uv_process_t *req, int64_t exit_status, int term_signal) {
3 fprintf(stderr, "Process exited with status %" PRId64 ", signal %d\n", exit_status, term_signal);
4 uv_close((uv_handle_t*) req, NULL);
プロセス終了後、プロセスウォッチャーを閉じることは、必須です。
プロセスパラメータの変更¶
子プロセスを起動する前に、uv_process_options_tのフィールドを使って実行環境を制御できます。
実行ディレクトリの変更¶
uv_process_options_t.cwdに、対象のディレクトリを設定します。
環境変数の設定¶
uv_process_options_t.envは、NULLで終端した文字列配列で、各文字列は、VAR=VALUE形式です。これを使ってプロセスの環境変数を設定します。NULLに設定すると、親（この）プロセスの環境を引き継ぎます。
オプションフラグ¶
uv_process_options_t.flagsに、次のフラグのビット単位ORを設定すると、子プロセスの動作が変わります。
UV_PROCESS_SETUID- 子プロセスの実行ユーザーIDを、uv_process_options_t.uidに設定します。UV_PROCESS_SETGID- 子プロセスの実行グループIDを、uv_process_options_t.gidに設定します。
UID/GIDの変更はUnixでのみサポートされています。Windowsでは、uv_spawnは、UV_ENOTSUPで失敗します。
UV_PROCESS_WINDOWS_VERBATIM_ARGUMENTS- Windowsでは、uv_process_options_t.argsの引用やエスケープを行いません。Unixでは無視されます。UV_PROCESS_DETACHED- 子プロセスを新しいセッションで開始し、親プロセスが終了した後も動作し続けるようにします。以下の例を参照してください。
プロセスの切り離し¶
フラグ、UV_PROCESS_DETACHEDを渡すと、デーモンや、親から独立した子プロセスを起動できます。親が終了しても影響を受けません。
detach/main.c
1int main() {
2 loop = uv_default_loop();
3
4 char* args[3];
5 args[0] = "sleep";
6 args[1] = "100";
7 args[2] = NULL;
8
9 options.exit_cb = NULL;
10 options.file = "sleep";
11 options.args = args;
12 options.flags = UV_PROCESS_DETACHED;
13
14 int r;
15 if ((r = uv_spawn(loop, &child_req, &options))) {
16 fprintf(stderr, "%s\n", uv_strerror(r));
17 return 1;
18 }
19 fprintf(stderr, "Launched sleep with PID %d\n", child_req.pid);
20 uv_unref((uv_handle_t*) &child_req);
21
22 return uv_run(loop, UV_RUN_DEFAULT);
ただし、ハンドルは引き続き子プロセスを監視しているため、プログラムは終了しないことに注意してください。uv_unref()を使うと、より、起動後は待たずに済ませる形にできます。
プロセスへのシグナル送信¶
libuvは、Unixの標準の、kill(2)システムコールをラップし、Windowsでも同様の意味を持つ機能を実装しています。ただし、注意点が1つあります。SIGTERM、SIGINT、SIGKILLは、すべてプロセスを終了させます。uv_killのシグネチャは、次のとおりです。
uv_err_t uv_kill(int pid, int signum);
libuvで開始したプロセスには、代わりに、uv_process_killを使えます。これは、pidではなく、uv_process_tウォッチャーを第1引数として受け取ります。この場合、終了コールバックが呼び出された後に、ウォッチャーに対して、必ず uv_closeを呼び出してください。
シグナル¶
libuvは、Unixのシグナルのラッパーを提供し、Windowsにも部分的に対応しています。
uv_signal_init()でハンドルを初期化し、ループに関連付けます。そのハンドルで特定のシグナルを待ち受けるには、ハンドラ関数を渡して、uv_signal_start()を使います。各ハンドラに関連付けられるシグナル番号は1つだけで、その後の、uv_signal_start()の呼び出しは、前の関連付けを上書きします。uv_signal_stop()で監視を停止します。さまざまな使い方を示す小さな例を挙げます。
signal/main.c
1#include <stdio.h>
2#include <stdlib.h>
3#include <unistd.h>
4#include <uv.h>
5
6uv_loop_t* create_loop()
7{
8 uv_loop_t *loop = malloc(sizeof(uv_loop_t));
9 if (loop) {
10 uv_loop_init(loop);
11 }
12 return loop;
13}
14
15void signal_handler(uv_signal_t *handle, int signum)
16{
17 printf("Signal received: %d\n", signum);
18 uv_signal_stop(handle);
19}
20
21// two signal handlers in one loop
22void thread1_worker(void *userp)
23{
24 uv_loop_t *loop1 = create_loop();
25
26 uv_signal_t sig1a, sig1b;
27 uv_signal_init(loop1, &sig1a);
28 uv_signal_start(&sig1a, signal_handler, SIGUSR1);
29
30 uv_signal_init(loop1, &sig1b);
31 uv_signal_start(&sig1b, signal_handler, SIGUSR1);
32
33 uv_run(loop1, UV_RUN_DEFAULT);
34}
35
36// two signal handlers, each in its own loop
37void thread2_worker(void *userp)
38{
39 uv_loop_t *loop2 = create_loop();
40 uv_loop_t *loop3 = create_loop();
41
42 uv_signal_t sig2;
43 uv_signal_init(loop2, &sig2);
44 uv_signal_start(&sig2, signal_handler, SIGUSR1);
45
46 uv_signal_t sig3;
47 uv_signal_init(loop3, &sig3);
48 uv_signal_start(&sig3, signal_handler, SIGUSR1);
49
50 while (uv_run(loop2, UV_RUN_NOWAIT) || uv_run(loop3, UV_RUN_NOWAIT)) {
51 }
52}
53
54int main()
55{
56 printf("PID %d\n", getpid());
57
58 uv_thread_t thread1, thread2;
59
60 uv_thread_create(&thread1, thread1_worker, 0);
61 uv_thread_create(&thread2, thread2_worker, 0);
62
63 uv_thread_join(&thread1);
64 uv_thread_join(&thread2);
65 return 0;
66}
注記
uv_run(loop, UV_RUN_NOWAIT)は、uv_run(loop, UV_RUN_ONCE)と同様に、1つのイベントだけを処理します。UV_RUN_ONCEは、保留中のイベントがなければブロックしますが、UV_RUN_NOWAITは直ちに戻ります。ここではNOWAITを使い、一方のループに保留中の活動がないために、もう一方のループが実行機会を失わないようにしています。
プロセスに、SIGUSR1を送ると、uv_signal_tごとに1回、計4回ハンドラが呼び出されます。ハンドラは、各ハンドルを停止するだけなので、プログラムは終了します。すべてのハンドラへのこのような通知は、非常に役立ちます。複数のイベントループを使うサーバーでは、各ループに、SIGINTのウォッチャーを追加するだけで、終了前にすべてのデータを安全に保存するようにできます。
子プロセスのI/O¶
通常、新しく起動されたプロセスには独自のファイル記述子があり、0、1、2は、それぞれ、stdin、stdout、stderrです。子プロセスとファイル記述子を共有したい場合もあります。たとえば、アプリケーションがサブコマンドを起動し、エラーはログファイルに入れたい一方、stdoutは無視したいとします。その場合、子プロセスの、stderrを、親のstderrと同じものにしたいでしょう。libuvは、このようにファイル記述子を、継承することをサポートしています。この例では、次のテストプログラムを呼び出します。
proc-streams/test.c
#include <stdio.h>
int main()
{
fprintf(stderr, "This is stderr\n");
printf("This is stdout\n");
return 0;
}
実際のプログラム、proc-streamsは、stderrだけを共有して、このテストを実行します。子プロセスのファイル記述子は、stdioフィールド（uv_process_options_t）で設定します。まず、stdio_countフィールドに、設定するファイル記述子の数を設定します。uv_process_options_t.stdioは、uv_stdio_container_tの配列で、次のようになっています。
typedef struct uv_stdio_container_s {
uv_stdio_flags flags;
union {
uv_stream_t* stream;
int fd;
} data;
} uv_stdio_container_t;
flagsには、いくつかの値を指定できます。使用しない場合は、UV_IGNOREを使います。最初の3つの、stdioフィールドを、UV_IGNOREにすると、/dev/nullへリダイレクトされます。
既存の記述子を渡したいので、UV_INHERIT_FDを使います。次に、fdを、stderrに設定します。
proc-streams/main.c
1
2int main() {
3 loop = uv_default_loop();
4
5 /* ... */
6
7 options.stdio_count = 3;
8 uv_stdio_container_t child_stdio[3];
9 child_stdio[0].flags = UV_IGNORE;
10 child_stdio[1].flags = UV_IGNORE;
11 child_stdio[2].flags = UV_INHERIT_FD;
12 child_stdio[2].data.fd = 2;
13 options.stdio = child_stdio;
14
15 options.exit_cb = on_exit;
16 options.file = args[0];
17 options.args = args;
18
19 int r;
20 if ((r = uv_spawn(loop, &child_req, &options))) {
21 fprintf(stderr, "%s\n", uv_strerror(r));
22 return 1;
23 }
24
25 return uv_run(loop, UV_RUN_DEFAULT);
26}
proc-streamを実行すると、「This is stderr」という行だけが表示されます。stdoutも継承するようにして、出力を確認してください。
このリダイレクトをストリームに適用するのは、とても簡単です。flagsを、UV_INHERIT_STREAMに設定し、data.streamに親プロセスのストリームを設定すると、子プロセスはそのストリームを標準I/Oとして扱えます。これを使って、CGIのようなものを実装できます。
CGIスクリプト・実行ファイルの例を示します。
cgi/tick.c
#include <stdio.h>
#include <unistd.h>
int main() {
int i;
for (i = 0; i < 10; i++) {
printf("tick\n");
fflush(stdout);
sleep(1);
}
printf("BOOM!\n");
return 0;
}
CGIサーバーでは、この章と、ネットワークの概念を組み合わせ、各クライアントに10回のtickを送った後、その接続を閉じます。
cgi/main.c
1
2void on_new_connection(uv_stream_t *server, int status) {
3 if (status == -1) {
4 // error!
5 return;
6 }
7
8 uv_tcp_t *client = (uv_tcp_t*) malloc(sizeof(uv_tcp_t));
9 uv_tcp_init(loop, client);
10 if (uv_accept(server, (uv_stream_t*) client) == 0) {
11 invoke_cgi_script(client);
12 }
13 else {
14 uv_close((uv_handle_t*) client, NULL);
15 }
ここでは単にTCP接続を受け入れ、ソケット（ストリーム）を、invoke_cgi_scriptに渡します。
cgi/main.c
1
2 args[1] = NULL;
3
4 /* ... finding the executable path and setting up arguments ... */
5
6 options.stdio_count = 3;
7 uv_stdio_container_t child_stdio[3];
8 child_stdio[0].flags = UV_IGNORE;
9 child_stdio[1].flags = UV_INHERIT_STREAM;
10 child_stdio[1].data.stream = (uv_stream_t*) client;
11 child_stdio[2].flags = UV_IGNORE;
12 options.stdio = child_stdio;
13
14 options.exit_cb = cleanup_handles;
15 options.file = args[0];
16 options.args = args;
17
18 // Set this so we can close the socket after the child process exits.
19 child_req.data = (void*) client;
20 int r;
21 if ((r = uv_spawn(loop, &child_req, &options))) {
22 fprintf(stderr, "%s\n", uv_strerror(r));
CGIスクリプトの、stdoutをソケットに設定するため、tickスクリプトが出力したものは、すべてクライアントに送られます。プロセスを使うと、読み書きのバッファリングをオペレーティングシステムに任せられるので、とても便利です。ただし、プロセスの作成は高コストな処理であることに注意してください。
親子間IPC¶
親と子は、uv_stdio_container_t.flagsに、UV_CREATE_PIPEと、UV_READABLE_PIPEまたは、UV_WRITABLE_PIPEをビット単位で組み合わせて設定して作るパイプで、単方向または双方向に通信できます。読み書きフラグは、子プロセスの視点で指定します。この場合、uv_stream_t* streamフィールドは、初期化済みで、まだ開いていない、uv_pipe_tインスタンスを指すように設定する必要があります。
新しいstdioパイプ¶
uv_pipe_t構造体は、単に、pipe(7)（または、|）だけを表すものではなく、ストリームとして扱えるファイル状のオブジェクト全般をサポートします。Windowsでは、この条件に当てはまるのは、名前付きパイプだけです。Unixでは、Unixドメインソケット、mkfifo(1)によるもの、あるいは、実際の、pipe(7)などです。uv_spawnが、uv_pipe_tを、UV_CREATE_PIPEフラグに基づいて初期化する場合は、socketpair(2)の作成を選びます。
これは、複数のlibuvプロセスがIPCで通信できるようにするためのものです。以下で説明します。
任意のプロセス間のIPC¶
ドメインソケットは、[1]既知の名前と、ファイルシステム内の場所を持てるので、無関係なプロセス間のIPCに使えます。オープンソースのデスクトップ環境で使われる、D-BUSシステムは、イベント通知にドメインソケットを使います。これにより、連絡先がオンラインになったときや、新しいハードウェアが検出されたときに、さまざまなアプリケーションが反応できます。MySQLサーバーも、クライアントが対話できるドメインソケットを動かしています。
ドメインソケットを使う場合、通常は、ソケットの作成者・所有者がサーバーになるクライアント・サーバーパターンに従います。初期設定が済めば、メッセージのやり取りはTCPと変わらないため、エコーサーバーの例を再利用します。
pipe-echo-server/main.c
1void remove_sock(int sig) {
2 uv_fs_t req;
3 uv_fs_unlink(loop, &req, PIPENAME, NULL);
4 exit(0);
5}
6
7int main() {
8 loop = uv_default_loop();
9
10 uv_pipe_t server;
11 uv_pipe_init(loop, &server, 0);
12
13 signal(SIGINT, remove_sock);
14
15 int r;
16 if ((r = uv_pipe_bind(&server, PIPENAME))) {
17 fprintf(stderr, "Bind error %s\n", uv_err_name(r));
18 return 1;
19 }
20 if ((r = uv_listen((uv_stream_t*) &server, 128, on_new_connection))) {
21 fprintf(stderr, "Listen error %s\n", uv_err_name(r));
22 return 2;
23 }
24 return uv_run(loop, UV_RUN_DEFAULT);
25}
ソケットには、echo.sockという名前を付け、ローカルディレクトリに作成します。ストリームAPIから見れば、このソケットはTCPソケットと同じように振る舞います。このサーバーは、socatでテストできます。
$ socat - /path/to/socket
ドメインソケットに接続するクライアントは、次を使います。
void uv_pipe_connect(uv_connect_t *req, uv_pipe_t *handle, const char *name, uv_connect_cb cb);
ここで、nameは、echo.sockなどです。Unixシステムでは、nameは、有効なファイル（たとえば、/tmp/echo.sock）を指す必要があります。Windowsでは、nameは、\\?\pipe\echo.sock形式に従います。
パイプでのファイル記述子の送信¶
ドメインソケットの便利な点は、ドメインソケットを通じてファイル記述子を送ることで、プロセス間でそれらを交換できることです。これにより、プロセスはI/Oを他のプロセスへ引き継げます。用途には、負荷分散サーバー、ワーカープロセス、その他のCPUを最適に活用する方法があります。現時点でlibuvがパイプを通じて送信できるのは、TCPソケットまたは他のパイプだけです。
実演として、クライアントをラウンドロビン方式でワーカープロセスに引き渡すエコーサーバーの実装を見てみます。このプログラムはやや複雑で、本書には抜粋だけを載せています。十分に理解するには、コード全体を読むことを勧めます。
ワーカープロセスは、マスタープロセスからファイル記述子を渡されるので、とても単純です。
multi-echo-server/worker.c
1
2uv_loop_t *loop;
3uv_pipe_t queue;
4int main() {
5 loop = uv_default_loop();
6
7 uv_pipe_init(loop, &queue, 1 /* ipc */);
8 uv_pipe_open(&queue, 0);
9 uv_read_start((uv_stream_t*)&queue, alloc_buffer, on_new_connection);
10 return uv_run(loop, UV_RUN_DEFAULT);
11}
queueは、反対側のマスタープロセスに接続されたパイプで、新しいファイル記述子が送られてきます。このパイプがプロセス間通信に使われることを示すため、ipc引数、つまり、uv_pipe_initの引数を1に設定することが重要です。マスタープロセスは、ワーカーの標準入力にファイルハンドルを書き込むので、パイプを、stdinに、uv_pipe_openで接続します。
multi-echo-server/worker.c
1void on_new_connection(uv_stream_t *q, ssize_t nread, const uv_buf_t *buf) {
2 if (nread < 0) {
3 if (nread != UV_EOF)
4 fprintf(stderr, "Read error %s\n", uv_err_name(nread));
5 uv_close((uv_handle_t*) q, NULL);
6 return;
7 }
8
9 uv_pipe_t *pipe = (uv_pipe_t*) q;
10 if (!uv_pipe_pending_count(pipe)) {
11 fprintf(stderr, "No pending count\n");
12 return;
13 }
14
15 uv_handle_type pending = uv_pipe_pending_type(pipe);
16 assert(pending == UV_TCP);
17
18 uv_tcp_t *client = (uv_tcp_t*) malloc(sizeof(uv_tcp_t));
19 uv_tcp_init(loop, client);
20 if (uv_accept(q, (uv_stream_t*) client) == 0) {
21 uv_os_fd_t fd;
22 uv_fileno((const uv_handle_t*) client, &fd);
23 fprintf(stderr, "Worker %d: Accepted fd %d\n", getpid(), fd);
24 uv_read_start((uv_stream_t*) client, alloc_buffer, echo_read);
25 }
26 else {
27 uv_close((uv_handle_t*) client, NULL);
28 }
29}
まず、uv_pipe_pending_count()で、読み出せるハンドルがあることを確認します。異なる種類のハンドルを扱うプログラムでは、uv_pipe_pending_type()で種類を判定できます。このコードの、acceptは奇妙に見えるかもしれませんが、実際には筋が通っています。通常、acceptが行うのは、別のファイル記述子（リスニングソケット）から、ファイル記述子（クライアント）を取得することです。ここで行うことも、まさに同じです。ファイル記述子（client）を、queueから取得します。ここから先、ワーカーは標準的なエコーサーバーの処理を行います。
次に、マスタープロセスに移り、負荷分散のためにワーカーをどのように起動するかを見てみましょう。
multi-echo-server/main.c
1struct child_worker {
2 uv_process_t req;
3 uv_process_options_t options;
4 uv_pipe_t pipe;
5} *workers;
child_worker構造体は、プロセスと、マスター・個々のプロセス間のパイプをまとめます。
multi-echo-server/main.c
1void setup_workers() {
2 round_robin_counter = 0;
3
4 // ...
5
6 // launch same number of workers as number of CPUs
7 uv_cpu_info_t *info;
8 int cpu_count;
9 uv_cpu_info(&info, &cpu_count);
10 uv_free_cpu_info(info, cpu_count);
11
12 child_worker_count = cpu_count;
13
14 workers = calloc(cpu_count, sizeof(struct child_worker));
15 while (cpu_count--) {
16 struct child_worker *worker = &workers[cpu_count];
17 uv_pipe_init(loop, &worker->pipe, 1);
18
19 uv_stdio_container_t child_stdio[3];
20 child_stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE;
21 child_stdio[0].data.stream = (uv_stream_t*) &worker->pipe;
22 child_stdio[1].flags = UV_IGNORE;
23 child_stdio[2].flags = UV_INHERIT_FD;
24 child_stdio[2].data.fd = 2;
25
26 worker->options.stdio = child_stdio;
27 worker->options.stdio_count = 3;
28
29 worker->options.exit_cb = close_process_handle;
30 worker->options.file = args[0];
31 worker->options.args = args;
32
33 uv_spawn(loop, &worker->req, &worker->options);
34 fprintf(stderr, "Started worker %d\n", worker->req.pid);
35 }
36}
ワーカーの設定では、便利なlibuv関数、uv_cpu_infoでCPU数を取得し、同じ数のワーカーを起動できるようにします。ここでも、IPCチャネルになるパイプの初期化時に、第3引数を1にすることが重要です。次に、子プロセスの、stdinを、読み取り可能なパイプ（子プロセスの視点で）にすることを示します。ここまでは単純です。ワーカーは起動され、標準入力にファイル記述子が書き込まれるのを待っています。
クライアントソケットを受け入れ、ラウンドロビン方式で次のワーカーに渡すのは、on_new_connectionです（TCPの基盤は、main()で初期化します）。
multi-echo-server/main.c
1void on_new_connection(uv_stream_t *server, int status) {
2 if (status == -1) {
3 // error!
4 return;
5 }
6
7 uv_tcp_t *client = (uv_tcp_t*) malloc(sizeof(uv_tcp_t));
8 uv_tcp_init(loop, client);
9 if (uv_accept(server, (uv_stream_t*) client) == 0) {
10 uv_write_t *write_req = (uv_write_t*) malloc(sizeof(uv_write_t));
11 dummy_buf = uv_buf_init("a", 1);
12 struct child_worker *worker = &workers[round_robin_counter];
13 uv_write2(write_req, (uv_stream_t*) &worker->pipe, &dummy_buf, 1, (uv_stream_t*) client, NULL);
14 round_robin_counter = (round_robin_counter + 1) % child_worker_count;
15 }
16 else {
17 uv_close((uv_handle_t*) client, NULL);
18 }
19}
uv_write2の呼び出しがすべての抽象化を処理するので、ハンドル（client）を適切な引数として渡すだけです。これで、マルチプロセスのエコーサーバーが動作します。
指摘してくれたKyleに感謝します。uv_write2()では、ハンドルを送る場合でも、空ではないバッファが必要です。