スレッド¶

基本的なスレッド操作¶

操作は多くありません。uv_thread_create()でスレッドを開始し、uv_thread_join()で終了を待つだけです。

thread-create/main.c

 1    int tracklen = 10;
 2    uv_thread_t hare_id;
 3    uv_thread_t tortoise_id;
 4    uv_thread_create(&hare_id, hare, &tracklen);
 5    uv_thread_create(&tortoise_id, tortoise, &tracklen);
 6
 7    uv_thread_join(&hare_id);
 8    uv_thread_join(&tortoise_id);
 9    return 0;
10}

第2引数はスレッドの入口になる関数で、最後の引数は、スレッドに任意のパラメータを渡すためのvoid *引数です。関数hareは、別のスレッドで、オペレーティングシステムによりプリエンプティブにスケジュールされて動作します。

thread-create/main.c

1    int tracklen = *((int *) arg);
2    while (tracklen) {
3        tracklen--;
4        uv_sleep(1000);
5        fprintf(stderr, "Hare ran another step\n");
6    }
7    fprintf(stderr, "Hare done running!\n");
8}
9

pthread_join()は、第2引数を使って、対象スレッドから呼び出し元のスレッドに値を返せますが、uv_thread_join()にはその機能がありません。値を送るには、スレッド間通信を使ってください。

同期プリミティブ¶

この節は意図的に簡潔にしています。本書はスレッドを主題としていないため、ここではlibuv APIで意外に感じる点だけを挙げます。それ以外は、pthreads(7)のmanページを参照してください。

int uv_mutex_init(uv_mutex_t* handle);
int uv_mutex_init_recursive(uv_mutex_t* handle);
void uv_mutex_destroy(uv_mutex_t* handle);
void uv_mutex_lock(uv_mutex_t* handle);
int uv_mutex_trylock(uv_mutex_t* handle);
void uv_mutex_unlock(uv_mutex_t* handle);

uv_mutex_init(a_mutex);
uv_mutex_lock(a_mutex);
uv_thread_create(thread_id, entry, (void *)a_mutex);
uv_mutex_lock(a_mutex);
// more things here

 1#include <stdio.h>
 2#include <uv.h>
 3
 4uv_barrier_t blocker;
 5uv_rwlock_t numlock;
 6int shared_num;
 7
 8void reader(void *n)
 9{
10    int num = *(int *)n;
11    int i;
12    for (i = 0; i < 20; i++) {
13        uv_rwlock_rdlock(&numlock);
14        printf("Reader %d: acquired lock\n", num);
15        printf("Reader %d: shared num = %d\n", num, shared_num);
16        uv_rwlock_rdunlock(&numlock);
17        printf("Reader %d: released lock\n", num);
18    }
19    uv_barrier_wait(&blocker);
20}
21
22void writer(void *n)
23{
24    int num = *(int *)n;
25    int i;
26    for (i = 0; i < 20; i++) {
27        uv_rwlock_wrlock(&numlock);
28        printf("Writer %d: acquired lock\n", num);
29        shared_num++;
30        printf("Writer %d: incremented shared num = %d\n", num, shared_num);
31        uv_rwlock_wrunlock(&numlock);
32        printf("Writer %d: released lock\n", num);
33    }
34    uv_barrier_wait(&blocker);
35}
36
37int main()
38{
39    uv_barrier_init(&blocker, 4);
40
41    shared_num = 0;
42    uv_rwlock_init(&numlock);
43
44    uv_thread_t threads[3];
45
46    int thread_nums[] = {1, 2, 1};
47    uv_thread_create(&threads[0], reader, &thread_nums[0]);
48    uv_thread_create(&threads[1], reader, &thread_nums[1]);
49
50    uv_thread_create(&threads[2], writer, &thread_nums[2]);
51
52    uv_barrier_wait(&blocker);
53    uv_barrier_destroy(&blocker);
54
55    uv_rwlock_destroy(&numlock);
56    return 0;
57}

/* Initialize guard */
static uv_once_t once_only = UV_ONCE_INIT;

int i = 0;

void increment() {
    i++;
}

void thread1() {
    /* ... work */
    uv_once(once_only, increment);
}

void thread2() {
    /* ... work */
    uv_once(once_only, increment);
}

int main() {
    /* ... spawn threads */
}

libuvのワークキュー¶

uv_queue_work()は、アプリケーションが別のスレッドでタスクを実行し、タスクの完了時にコールバックを呼び出せるようにする便利な関数です。一見単純な関数ですが、uv_queue_work()の魅力は、さまざまなサードパーティライブラリをイベントループ方式で使えるようになる点です。イベントループを使う場合、ループのスレッドで定期的に動く関数が、I/Oでブロックしたり、大量のCPU時間を消費したりしないことを、必ず確認しなければなりません。そうでなければループが遅くなり、イベントを十分な能力で処理できなくなるためです。

しかし、既存のコードには、ブロッキング関数（たとえば内部でI/Oを行うルーチン）を含むものが多くあります。応答性を確保するには、スレッドで使う必要があるコードです（従来の「クライアントごとに1スレッド」というサーバーモデル）。これらをイベントループライブラリと組み合わせるには、通常、タスクを別のスレッドで実行する独自の仕組みを作る必要があります。libuvは、そのための便利な抽象化を提供します。

次は、node.js is cancerに着想を得た単純な例です。途中で少しスリープしながらフィボナッチ数を計算します。ただし、別のスレッドで実行するので、このブロッキングかつCPU負荷の高いタスクが、イベントループの他の活動を妨げません。

queue-work/main.c - のんびりしたフィボナッチ計算

 1void fib(uv_work_t *req) {
 2    int n = *(int *) req->data;
 3    if (random() % 2)
 4        sleep(1);
 5    else
 6        sleep(3);
 7    long fib = fib_(n);
 8    fprintf(stderr, "%dth fibonacci is %lu\n", n, fib);
 9}
10
11void after_fib(uv_work_t *req, int status) {
12    fprintf(stderr, "Done calculating %dth fibonacci\n", *(int *) req->data);
13}

実際のタスク関数は単純で、別のスレッドで動くことを示すものはありません。手掛かりは、uv_work_t構造体です。そのvoid* dataフィールドで任意のデータを渡し、スレッドとの間でやり取りできます。ただし、両方のスレッドが動作している間に値を変更する場合は、適切なロックを使ってください。

起動する関数は、uv_queue_workです。

queue-work/main.c

 1int main() {
 2    loop = uv_default_loop();
 3
 4    int data[FIB_UNTIL];
 5    uv_work_t req[FIB_UNTIL];
 6    int i;
 7    for (i = 0; i < FIB_UNTIL; i++) {
 8        data[i] = i;
 9        req[i].data = (void *) &data[i];
10        uv_queue_work(loop, &req[i], fib, after_fib);
11    }
12
13    return uv_run(loop, UV_RUN_DEFAULT);
14}

スレッド関数は別のスレッドで起動され、uv_work_t構造体を受け取ります。関数が戻ると、after関数が、イベントループを実行しているスレッドで呼び出されます。同じ構造体が渡されます。

ブロッキングライブラリのラッパーを書く際によく使われるパターンは、バトンを使ったデータの受け渡しです。

libuvのバージョン0.9.4以降では、追加の関数、uv_cancel()が利用できます。これにより、libuvのワークキュー上のタスクをキャンセルできます。キャンセルできるのは、まだ開始していないタスクだけです。タスクが、すでに実行を開始しているか、実行を終了している場合、uv_cancel() は失敗します。

uv_cancel()は、ユーザーが終了を要求したときに、保留中のタスクを片付けるのに役立ちます。たとえば、音楽プレーヤーは、音声ファイルを探すために複数のディレクトリをキューに入れることがあります。ユーザーがプログラムを終了したら、保留中のすべての要求が実行されるのを待たずに、すぐ終了すべきです。

フィボナッチの例を変更して、uv_cancel()を実演しましょう。まず、終了用のシグナルハンドラを設定します。

queue-cancel/main.c

 1int main() {
 2    loop = uv_default_loop();
 3
 4    int data[FIB_UNTIL];
 5    int i;
 6    for (i = 0; i < FIB_UNTIL; i++) {
 7        data[i] = i;
 8        fib_reqs[i].data = (void *) &data[i];
 9        uv_queue_work(loop, &fib_reqs[i], fib, after_fib);
10    }
11
12    uv_signal_t sig;
13    uv_signal_init(loop, &sig);
14    uv_signal_start(&sig, signal_handler, SIGINT);
15
16    return uv_run(loop, UV_RUN_DEFAULT);
17}

ユーザーが、Ctrl+Cを押してシグナルを発生させると、uv_cancel()をすべてのワーカーに送ります。uv_cancel()は、実行待ちでキャンセルできたタスクに対して、0を返します。

queue-cancel/main.c

1void signal_handler(uv_signal_t *req, int signum)
2{
3    printf("Signal received!\n");
4    int i;
5    for (i = 0; i < FIB_UNTIL; i++) {
6        uv_cancel((uv_req_t*) &fib_reqs[i]);
7    }
8    uv_signal_stop(req);
9}

キャンセルに成功したタスクでは、after関数が、statusを、UV_ECANCELEDに設定して呼び出されます。

queue-cancel/main.c

1void after_fib(uv_work_t *req, int status) {
2    if (status == UV_ECANCELED)
3        fprintf(stderr, "Calculation of %d cancelled.\n", *(int *) req->data);
4}

uv_cancel()は、uv_fs_t要求と、uv_getaddrinfo_t要求にも使えます。ファイルシステム系の関数では、uv_fs_t.errornoが、UV_ECANCELEDに設定されます。

スレッド間通信¶

複数のスレッドに、互いにメッセージを送り合ってほしい場合があります。たとえば、動作中の別のスレッドで、長時間のタスク（uv_queue_workを使う場合など）を実行しつつ、メインスレッドに進捗を通知したい場合です。ここでは、ダウンロードマネージャーが、実行中のダウンロードの状態をユーザーに知らせる単純な例を示します。

progress/main.c

 1
 2uv_loop_t *loop;
 3}
 4
 5
 6void print_progress(uv_async_t *handle) {
 7    double pct = atomic_load_explicit(&percentage, memory_order_acquire);
 8    fprintf(stderr, "Downloaded %.2f%%\n", pct);
 9}
10
11int main() {
12    loop = uv_default_loop();
13
14    uv_work_t req;
15    int size = 10240;
16    req.data = (void*) &size;
17
18    uv_async_init(loop, &async, print_progress);
19    uv_queue_work(loop, &req, fake_download, after);
20    return uv_run(loop, UV_RUN_DEFAULT);
21}

非同期のスレッド間通信は、ループ上で動作します。そのため、どのスレッドでもメッセージの送信者になれますが、受信者になれるのはlibuvのループを持つスレッドだけです（正確には、受信者はループです）。libuvはメッセージを受け取ると、非同期ウォッチャーを渡してコールバック（print_progress）を呼び出します。

progress/main.c

 1
 2_Atomic double percentage;
 3
 4void fake_download(uv_work_t *req) {
 5    int size = *((int*) req->data);
 6    int downloaded = 0;
 7    double pct;
 8    while (downloaded < size) {
 9        pct = downloaded * 100.0 / size;
10        atomic_store_explicit(&percentage, pct, memory_order_release);
11        uv_async_send(&async);
12
13        sleep(1);
14        downloaded += (200+random())%1000; // can only download max 1000bytes/sec,
15                                           // but at least a 200;

ダウンロード関数では、進捗表示を更新し、uv_async_sendで配信用のメッセージをキューに入れます。注意してください。uv_async_sendもノンブロッキングで、直ちに戻ります。

progress/main.c

1
2void after(uv_work_t *req, int status) {
3    fprintf(stderr, "Download complete\n");
4    uv_close((uv_handle_t*) &async, NULL);

コールバックは、ウォッチャーからデータを取り出す、標準的なlibuvのパターンです。

最後に、ウォッチャーを片付けることを忘れないでください。

progress/main.c

1    // Ensure final 100% progress update is sent
2    pct = 100.0;
3    atomic_store_explicit(&percentage, pct, memory_order_release);
4    uv_async_send(&async);

この例では、dataフィールドを不適切に使っていました。この例の後、bnoordhuisが、dataフィールドの使用はスレッドセーフではなく、uv_async_send()は実際にはイベントループを起こすためだけのものだと指摘しました。アクセスが適切な順序で行われるように、ミューテックスまたは読み書きロックを使ってください。

uv_async_sendが必要な用途の1つは、機能のために特定のスレッドとの結び付きが必要なライブラリと相互運用する場合です。たとえばnode.jsでは、v8エンジンのインスタンス、コンテキスト、そのオブジェクトは、v8インスタンスを開始したスレッドに結び付いています。別のスレッドからv8のデータ構造を操作すると、未定義の結果につながる可能性があります。ここで、サードパーティライブラリを結び付けるnode.jsモジュールを考えてみましょう。次のようになるかもしれません。

1. nodeでは、追加情報を得るために呼び出すJavaScriptコールバックを使って、サードパーティライブラリを設定します。

   var lib = require('lib');
   lib.on_progress(function() {
       console.log("Progress");
   });
   
   lib.do();
   
   // do other stuff
   

2. lib.doはノンブロッキングであるはずですが、サードパーティライブラリはブロッキングなので、バインディングでは、uv_queue_workを使います。

3. 実際の処理は別のスレッドで動作しており、進捗コールバックを呼びたいのですが、JavaScriptを操作するためにv8を直接呼び出すことはできません。そのため、uv_async_sendを使います。

4. メインループのスレッド、すなわちv8のスレッドで呼ばれる非同期コールバックが、v8とやり取りしてJavaScriptのコールバックを呼び出します。