Libuv polling外部IO

通常在使用第三方库的时候,需要应对他们自己的IO,还有保持监视他们的socket和内部文件。在此情形下,不可能使用标准的IO流操作,但第三方库仍然能整合进event-loop中。所有这些需要的就是,第三方库就必须允许你访问它的底层文件描述符,并且提供可以处理有用户定义的细微任务的函数。但是一些第三库并不允许你这么做,他们只提供了一个标准的阻塞IO函数,此函数会完成所有的工作并返回。在event-loop的线程直接使用它们是不明智的,而是应该使用libuv的工作线程。当然,这也意味着失去了对第三方库的颗粒化控制。

libuv的uv_poll简单地监视了使用了操作系统的监控机制的文件描述符。从某方面说,libuv实现的所有的IO操作,的背后均有uv_poll的支持。无论操作系统何时监视到文件描述符的改变,libuv都会调用响应的回调函数。

现在我们简单地实现一个下载管理程序,它会通过libcurl来下载文件。我们不会直接控制libcurl,而是使用libuv的event-loop,通过非阻塞的异步的多重接口来处理下载,与此同时,libuv会监控IO的就绪状态。

uvwget/main.c - The setup

#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <uv.h>
#include <curl/curl.h>

uv_loop_t *loop;
CURLM *curl_handle;
uv_timer_t timeout;
}

int main(int argc, char **argv) {
    loop = uv_default_loop();

    if (argc <= 1)
        return 0;

    if (curl_global_init(CURL_GLOBAL_ALL)) {
        fprintf(stderr, "Could not init cURL\n");
        return 1;
    }

    uv_timer_init(loop, &timeout);

    curl_handle = curl_multi_init();
    curl_multi_setopt(curl_handle, CURLMOPT_SOCKETFUNCTION, handle_socket);
    curl_multi_setopt(curl_handle, CURLMOPT_TIMERFUNCTION, start_timeout);

    while (argc-- > 1) {
        add_download(argv[argc], argc);
    }

    uv_run(loop, UV_RUN_DEFAULT);
    curl_multi_cleanup(curl_handle);
    return 0;
}

每种库整合进libuv的方式都是不同的。以libcurl的例子来说,我们注册了两个回调函数。socket回调函数handle_socket会在socket状态改变的时候被触发,因此我们不得不开始轮询它。start_timeout是libcurl用来告知我们下一次的超时间隔的,之后我们就应该不管当前IO状态,驱动libcurl向前。这些也就是libcurl能处理错误或驱动下载进度向前的原因。

可以这么调用下载器:

$ ./uvwget [url1] [url2] ...

我们可以把url当成参数传入程序。

uvwget/main.c - Adding urls
void add_download(const char *url, int num) {
    char filename[50];
    sprintf(filename, "%d.download", num);
    FILE *file;

    file = fopen(filename, "w");
    if (file == NULL) {
        fprintf(stderr, "Error opening %s\n", filename);
        return;
    }

    CURL *handle = curl_easy_init();
    curl_easy_setopt(handle, CURLOPT_WRITEDATA, file);
    curl_easy_setopt(handle, CURLOPT_URL, url);
    curl_multi_add_handle(curl_handle, handle);
    fprintf(stderr, "Added download %s -> %s\n", url, filename);
}

我们允许libcurl直接向文件写入数据。

start_timeout会被libcurl立即调用。它会启动一个libuv的定时器,使用CURL_SOCKET_TIMEOUT驱动curl_multi_socket_action,当其超时时,调用它。curl_multi_socket_action会驱动libcurl,也会在socket状态改变的时候被调用。但在我们深入讲解它之前,我们需要轮询监听socket,等待handle_socket被调用。

uvwget/main.c - Setting up polling

void start_timeout(CURLM *multi, long timeout_ms, void *userp) {
    if (timeout_ms <= 0)
        timeout_ms = 1; /* 0 means directly call socket_action, but we'll do it in a bit */
    uv_timer_start(&timeout, on_timeout, timeout_ms, 0);
}

int handle_socket(CURL *easy, curl_socket_t s, int action, void *userp, void *socketp) {
    curl_context_t *curl_context;
    if (action == CURL_POLL_IN || action == CURL_POLL_OUT) {
        if (socketp) {
            curl_context = (curl_context_t*) socketp;
        }
        else {
            curl_context = create_curl_context(s);
            curl_multi_assign(curl_handle, s, (void *) curl_context);
        }
    }

    switch (action) {
        case CURL_POLL_IN:
            uv_poll_start(&curl_context->poll_handle, UV_READABLE, curl_perform);
            break;
        case CURL_POLL_OUT:
            uv_poll_start(&curl_context->poll_handle, UV_WRITABLE, curl_perform);
            break;
        case CURL_POLL_REMOVE:
            if (socketp) {
                uv_poll_stop(&((curl_context_t*)socketp)->poll_handle);
                destroy_curl_context((curl_context_t*) socketp);                
                curl_multi_assign(curl_handle, s, NULL);
            }
            break;
        default:
            abort();
    }

    return 0;
}

我们关心的是socket的文件描述符s,还有action。对应每一个socket,我们都创造了uv_poll_t,并用curl_multi_assign把它们关联起来。每当回调函数被调用时,socketp都会指向它。

在下载完成或失败后,libcurl需要移除poll。所以我们停止并回收了poll的handle。

我们使用UV_READABLE或UV_WRITABLE开始轮询,基于libcurl想要监视的事件。当socket已经准备好读或写后,libuv会调用轮询的回调函数。在相同的handle上调用多次uv_poll_start是被允许的,这么做可以更新事件的参数。curl_perform是整个程序的关键。

uvwget/main.c - Driving libcurl.
void curl_perform(uv_poll_t *req, int status, int events) {
    uv_timer_stop(&timeout);
    int running_handles;
    int flags = 0;
    if (status < 0)                      flags = CURL_CSELECT_ERR;
    if (!status && events & UV_READABLE) flags |= CURL_CSELECT_IN;
    if (!status && events & UV_WRITABLE) flags |= CURL_CSELECT_OUT;

    curl_context_t *context;

    context = (curl_context_t*)req;

    curl_multi_socket_action(curl_handle, context->sockfd, flags, &running_handles);
    check_multi_info();   
}

首先我们要做的是停止定时器,因为内部还有其他要做的事。接下来我们我们依据触发回调函数的事件,来设置flag。然后,我们使用上述socket和flag作为参数,来调用curl_multi_socket_action。在此刻libcurl会在内部完成所有的工作,然后尽快地返回事件驱动程序在主线程中急需的数据。libcurl会在自己的队列中将传输进度的消息排队。对于我们来说,我们只关心是否传输完成,这类消息。所以我们将这类消息提取出来,并将传输完成的handle回收。

uvwget/main.c - Reading transfer status.

void check_multi_info(void) {
    char *done_url;
    CURLMsg *message;
    int pending;

    while ((message = curl_multi_info_read(curl_handle, &pending))) {
        switch (message->msg) {
        case CURLMSG_DONE:
            curl_easy_getinfo(message->easy_handle, CURLINFO_EFFECTIVE_URL,
                            &done_url);
            printf("%s DONE\n", done_url);

            curl_multi_remove_handle(curl_handle, message->easy_handle);
            curl_easy_cleanup(message->easy_handle);
            break;

        default:
            fprintf(stderr, "CURLMSG default\n");
            abort();
        }
    }
}