线程核心操作(Core thread operations)¶

下面的例子程序并没有太多代码, 只是通过 uv_thread_create 创建了线程, 然后调用 uv_thread_join 等待线程退出.

thread-create/main.c

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

int main() {
    int tracklen = 10;
    uv_thread_t hare_id;
    uv_thread_t tortoise_id;
    uv_thread_create(&hare_id, hare, &tracklen);
    uv_thread_create(&tortoise_id, tortoise, &tracklen);

    uv_thread_join(&hare_id);
    uv_thread_join(&tortoise_id);
    return 0;
}

第二个参数是线程执行任务的入口函数地址, 最后 void * 类型的参数用来向线程传递自定义参数. 函数 hare 现在可以在另外一个单独线程中运行, 并由操作系统进行调度.

thread-create/main.c

1
2
3
4
5
6
7
8
9

void hare(void *arg) {
    int tracklen = *((int *) arg);
    while (tracklen) {
        tracklen--;
        sleep(1);
        fprintf(stderr, "Hare ran another step\n");
    }
    fprintf(stderr, "Hare done running!\n");
}

与 pthread_join() 允许目标线程通过 pthread_join() 的第二个函数传回给调用线程一个返回值不同, uv_thread_join() 并不允许调用/被调用线程之间通过这样的方式传值. 你应该使用 线程间通信(Inter-thread communication).

同步原语(Synchronization Primitives)¶

本节为了简洁起见并不完整简述线程的有关内容, 我在此也只归纳出一些 libuv 线程 API 比较特殊的地方, 剩下的你可以参看 pthread 的帮助页面 pthreads.

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

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

#include <stdio.h>
#include <uv.h>

uv_barrier_t blocker;
uv_rwlock_t numlock;
int shared_num;

void reader(void *n)
{
    int num = *(int *)n;
    int i;
    for (i = 0; i < 20; i++) {
        uv_rwlock_rdlock(&numlock);
        printf("Reader %d: acquired lock\n", num);
        printf("Reader %d: shared num = %d\n", num, shared_num);
        uv_rwlock_rdunlock(&numlock);
        printf("Reader %d: released lock\n", num);
    }
    uv_barrier_wait(&blocker);
}

void writer(void *n)
{
    int num = *(int *)n;
    int i;
    for (i = 0; i < 20; i++) {
        uv_rwlock_wrlock(&numlock);
        printf("Writer %d: acquired lock\n", num);
        shared_num++;
        printf("Writer %d: incremented shared num = %d\n", num, shared_num);
        uv_rwlock_wrunlock(&numlock);
        printf("Writer %d: released lock\n", num);
    }
    uv_barrier_wait(&blocker);
}

int main()
{
    uv_barrier_init(&blocker, 4);

    shared_num = 0;
    uv_rwlock_init(&numlock);

    uv_thread_t threads[3];

    int thread_nums[] = {1, 2, 1};
    uv_thread_create(&threads[0], reader, &thread_nums[0]);
    uv_thread_create(&threads[1], reader, &thread_nums[1]);

    uv_thread_create(&threads[2], writer, &thread_nums[2]);

    uv_barrier_wait(&blocker);
    uv_barrier_destroy(&blocker);

    uv_rwlock_destroy(&numlock);
    return 0;
}

/* 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 任务), (典型的 ‘one thread per client’ 服务器模型), 并在单独的线程中运行某一任务. libuv 只是提供了一层抽象而已.

下面是一个简单的例子(原型为 node.js is cancer). 我们程序计算 fibonacci 数, 中途也会休眠一会,但是由于是在单独的线程中运行的, 所以阻塞和 CPU 计算密集型的任务(fibonacci 数的计算)并不会阻碍事件循环执行其它任务.

queue-work/main.c - lazy fibonacci

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

void fib(uv_work_t *req) {
    int n = *(int *) req->data;
    if (random() % 2)
        sleep(1);
    else
        sleep(3);
    long fib = fib_(n);
    fprintf(stderr, "%dth fibonacci is %lu\n", n, fib);
}

void after_fib(uv_work_t *req, int status) {
    fprintf(stderr, "Done calculating %dth fibonacci\n", *(int *) req->data);
}

真正的执行任务的函数比较简单, 只是在单独的线程中执行. uv_work_t 结构是线索, 你可以通过 void* date 传递任意数据, 并且利用该指针和执行线程通信(数据传递). 但注意, 如果在多个线程中同时修改某个变量, 就应该使用合适的锁来保护共享变量.

触发器 是 uv_queue_work:

queue-work/main.c

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

int main() {
    loop = uv_default_loop();

    int data[FIB_UNTIL];
    uv_work_t req[FIB_UNTIL];
    int i;
    for (i = 0; i < FIB_UNTIL; i++) {
        data[i] = i;
        req[i].data = (void *) &data[i];
        uv_queue_work(loop, &req[i], fib, after_fib);
    }

    return uv_run(loop, UV_RUN_DEFAULT);
}

线程函数会在单独的线程中被启动, 并传入 uv_work_t 结构, 一旦函数返回, 就会调用 after_fib 函数, 同时也传入 uv_work_t 结构的指针.

如果需要封装程序库的接口, 请参考 pattern 一节如何向线程传递数据.

线程间通信(Inter-thread communication)¶

某些时候你想在同时运行的各个线程之间传递数据. 例如你可能在一个单独的线程中运行一个长时间任务(可能是通过 uv_queue_work), 但是需要向主线程汇报任务进度. 一个简单的例子是下载管理器向用户汇报当前的下载任务的状态.

progress/main.c

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

uv_loop_t *loop;
uv_async_t async;

int main() {
    loop = uv_default_loop();

    uv_work_t req;
    int size = 10240;
    req.data = (void*) &size;

    uv_async_init(loop, &async, print_progress);
    uv_queue_work(loop, &req, fake_download, after);

    return uv_run(loop, UV_RUN_DEFAULT);
}

异步线程通信是在事件循环上进行的, 因此尽管任何线程都可能成为消息发送者, 但是只有 libuv 事件循环的线程可以是接收者(或者事件循环本身是接收者). 在任何时候异步监视器接受到了一个消息后, libuv 就会调用(print_progress)回调函数.

progress/main.c

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

void fake_download(uv_work_t *req) {
    int size = *((int*) req->data);
    int downloaded = 0;
    double percentage;
    while (downloaded < size) {
        percentage = downloaded*100.0/size;
        async.data = (void*) &percentage;
        uv_async_send(&async);

        sleep(1);
        downloaded += (200+random())%1000; // can only download max 1000bytes/sec,
                                           // but at least a 200;
    }
}

在下载函数中我们修改下载进度指示器, 并将消息写入 uv_async_send 发送队列, 记住: uv_async_send 也是非阻塞的, 它会立即返回.

progress/main.c

1
2
3
4

void print_progress(uv_async_t *handle, int status /*UNUSED*/) {
    double percentage = *((double*) handle->data);
    fprintf(stderr, "Downloaded %.2f%%\n", percentage);
}

回调函数从监视器中提取数据, 这也是 libuv 的标准模式.

最后应记得清除监视器.

progress/main.c

1
2
3
4

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

上面的例子滥用了 data 参数, bnoordhuis 指出使用 data 参数并不是线程安全的. uv_async_send() 实际上只会唤醒事件循环. 所以应该使用互斥量或者读写锁来保证对共享变量访问的顺序是正确的.

uv_async_send 的应用场景之一是不同功能的程序库之间需要线程级别的交互. 例如, 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. 在单独线程中完成的工作需要调用 progress 回调函数. 但是不能通过 JavaScript 直接访问 v8 引擎, 以使用 uv_async_send 与 v8 进行交互.

4. 在主线程(v8 线程)设置的异步回调函数就可以通过 JavaScript 回调函数和 v8 引擎进行通行.