int uv_fs_open(uv_loop_t* loop, uv_fs_t* req, const char* path, int flags, int mode, uv_fs_cb cb)

int uv_fs_close(uv_loop_t* loop, uv_fs_t* req, uv_file file, uv_fs_cb cb)

void callback(uv_fs_t* req);

void on_open(uv_fs_t *req) {if (req->result != -1) {uv_fs_read(uv_default_loop(), &read_req, req->result,buffer, sizeof(buffer), -1, on_read);}else {fprintf(stderr, "error opening file: %d\n", req->errorno);}uv_fs_req_cleanup(req);
}

必须调用 uv_fs_req_cleanup() 来释放 libuv 内部使用的内存空间.

void on_read(uv_fs_t *req) {uv_fs_req_cleanup(req);if (req->result < 0) {fprintf(stderr, "Read error: %s\n", uv_strerror(uv_last_error(uv_default_loop())));}else if (req->result == 0) {uv_fs_t close_req;// synchronousuv_fs_close(uv_default_loop(), &close_req, open_req.result, NULL);}else {uv_fs_write(uv_default_loop(), &write_req, 1, buffer, req->result, -1, on_write);}
}

void on_write(uv_fs_t *req) {uv_fs_req_cleanup(req);if (req->result < 0) {fprintf(stderr, "Write error: %s\n", uv_strerror(uv_last_error(uv_default_loop())));}else {uv_fs_read(uv_default_loop(), &read_req, open_req.result, buffer, sizeof(buffer), -1, on_read);}
}

错误值通常保存在 errno 并可以通过 uv_fs_t.errorno 获取, 但是被转换成了标准的UV_* 错误码. 目前还没有方法直接从 errorno 解析得到错误消息的字符串表示.

由于文件系统和磁盘通常为了提高性能吞吐率而配置了缓冲区, libuv 中一次 ‘成功’ 的写操作可能不会被立刻提交到磁盘上, 你可以通过uv_fs_fsync 来保证一致性.

int main(int argc, char **argv) {uv_fs_open(uv_default_loop(), &open_req, argv[1], O_RDONLY, 0, on_open);uv_run(uv_default_loop(), UV_RUN_DEFAULT);return 0;
}

UV_EXTERN int uv_fs_close(uv_loop_t* loop, uv_fs_t* req, uv_file file,uv_fs_cb cb);UV_EXTERN int uv_fs_open(uv_loop_t* loop, uv_fs_t* req, const char* path,int flags, int mode, uv_fs_cb cb);UV_EXTERN int uv_fs_read(uv_loop_t* loop, uv_fs_t* req, uv_file file,void* buf, size_t length, int64_t offset, uv_fs_cb cb);UV_EXTERN int uv_fs_unlink(uv_loop_t* loop, uv_fs_t* req, const char* path,uv_fs_cb cb);UV_EXTERN int uv_fs_write(uv_loop_t* loop, uv_fs_t* req, uv_file file,void* buf, size_t length, int64_t offset, uv_fs_cb cb);UV_EXTERN int uv_fs_mkdir(uv_loop_t* loop, uv_fs_t* req, const char* path,int mode, uv_fs_cb cb);UV_EXTERN int uv_fs_rmdir(uv_loop_t* loop, uv_fs_t* req, const char* path,uv_fs_cb cb);UV_EXTERN int uv_fs_readdir(uv_loop_t* loop, uv_fs_t* req,const char* path, int flags, uv_fs_cb cb);UV_EXTERN int uv_fs_stat(uv_loop_t* loop, uv_fs_t* req, const char* path,uv_fs_cb cb);UV_EXTERN int uv_fs_fstat(uv_loop_t* loop, uv_fs_t* req, uv_file file,uv_fs_cb cb);UV_EXTERN int uv_fs_rename(uv_loop_t* loop, uv_fs_t* req, const char* path,const char* new_path, uv_fs_cb cb);UV_EXTERN int uv_fs_fsync(uv_loop_t* loop, uv_fs_t* req, uv_file file,uv_fs_cb cb);UV_EXTERN int uv_fs_fdatasync(uv_loop_t* loop, uv_fs_t* req, uv_file file,uv_fs_cb cb);UV_EXTERN int uv_fs_ftruncate(uv_loop_t* loop, uv_fs_t* req, uv_file file,int64_t offset, uv_fs_cb cb);UV_EXTERN int uv_fs_sendfile(uv_loop_t* loop, uv_fs_t* req, uv_file out_fd,uv_file in_fd, int64_t in_offset, size_t length, uv_fs_cb cb);UV_EXTERN int uv_fs_chmod(uv_loop_t* loop, uv_fs_t* req, const char* path,int mode, uv_fs_cb cb);UV_EXTERN int uv_fs_utime(uv_loop_t* loop, uv_fs_t* req, const char* path,double atime, double mtime, uv_fs_cb cb);UV_EXTERN int uv_fs_futime(uv_loop_t* loop, uv_fs_t* req, uv_file file,double atime, double mtime, uv_fs_cb cb);UV_EXTERN int uv_fs_lstat(uv_loop_t* loop, uv_fs_t* req, const char* path,uv_fs_cb cb);UV_EXTERN int uv_fs_link(uv_loop_t* loop, uv_fs_t* req, const char* path,const char* new_path, uv_fs_cb cb);

int uv_read_start(uv_stream_t*, uv_alloc_cb alloc_cb, uv_read_cb read_cb);
int uv_read_stop(uv_stream_t*);
int uv_write(uv_write_t* req, uv_stream_t* handle,uv_buf_t bufs[], int bufcnt, uv_write_cb cb);

./uvtee <output_file>

int main(int argc, char **argv) {loop = uv_default_loop();uv_pipe_init(loop, &stdin_pipe, 0);uv_pipe_open(&stdin_pipe, 0);uv_pipe_init(loop, &stdout_pipe, 0);uv_pipe_open(&stdout_pipe, 1);uv_fs_t file_req;int fd = uv_fs_open(loop, &file_req, argv[1], O_CREAT | O_RDWR, 0644, NULL);uv_pipe_init(loop, &file_pipe, 0);uv_pipe_open(&file_pipe, fd);uv_read_start((uv_stream_t*)&stdin_pipe, alloc_buffer, read_stdin);uv_run(loop, UV_RUN_DEFAULT);return 0;
}

uv_buf_t alloc_buffer(uv_handle_t *handle, size_t suggested_size) {return uv_buf_init((char*) malloc(suggested_size), suggested_size);
}void read_stdin(uv_stream_t *stream, ssize_t nread, uv_buf_t buf) {if (nread == -1) {if (uv_last_error(loop).code == UV_EOF) {uv_close((uv_handle_t*)&stdin_pipe, NULL);uv_close((uv_handle_t*)&stdout_pipe, NULL);uv_close((uv_handle_t*)&file_pipe, NULL);}}else {if (nread > 0) {write_data((uv_stream_t*)&stdout_pipe, nread, buf, on_stdout_write);write_data((uv_stream_t*)&file_pipe, nread, buf, on_file_write);}}if (buf.base)free(buf.base);
}

typedef struct {uv_write_t req;uv_buf_t buf;
} write_req_t;void free_write_req(uv_write_t *req) {write_req_t *wr = (write_req_t*) req;free(wr->buf.base);free(wr);
}void on_stdout_write(uv_write_t *req, int status) {free_write_req(req);
}void on_file_write(uv_write_t *req, int status) {free_write_req(req);
}void write_data(uv_stream_t *dest, size_t size, uv_buf_t buf, uv_write_cb callback) {write_req_t *req = (write_req_t*) malloc(sizeof(write_req_t));req->buf = uv_buf_init((char*) malloc(size), size);memcpy(req->buf.base, buf.base, size);uv_write((uv_write_t*) req, (uv_stream_t*)dest, &req->buf, 1, callback);
}

TCP

TCP 是一种面向连接的流式协议, 因此是基于 libuv 的流式基础架构上的.

服务器(Server)

服务器端的 sockets 处理流程如下:

1. uv_tcp_init 初始化 TCP 监视器.
2. uv_tcp_bind 绑定.
3. 在指定的监视器上调用 uv_listen 来设置回调函数, 当有新的客户端连接到来时, libuv 就会调用设置的回调函数.
4. uv_accept 接受连接.
5. 使用 stream operations 与客户端进行通信.

以下是一个简单的 echo 服务器的例子:

int main() {loop = uv_default_loop();uv_tcp_t server;uv_tcp_init(loop, &server);struct sockaddr_in bind_addr = uv_ip4_addr("0.0.0.0", 7000);uv_tcp_bind(&server, bind_addr);int r = uv_listen((uv_stream_t*) &server, 128, on_new_connection);if (r) {fprintf(stderr, "Listen error %s\n", uv_err_name(uv_last_error(loop)));return 1;}return uv_run(loop, UV_RUN_DEFAULT);
}

你可以看到辅助函数 uv_ip4_addr 用来将人为可读的字符串类型的 IP 地址和端口号转换成 BSD 套接字 API 所需要的struct sockaddr_in 类型的结构. 逆变换可以使用uv_ip4_name 来完成.

对于 IPv6 来说应该使用 uv_ip6_* 形式的函数.

大部分的设置(setup)函数都是普通函数, 因为他们都是 计算密集型(CPU-bound), 直到调用了uv_listen 我们才回到 libuv 中回调函数风格.uv_listen 的第二个参数 backlog 队列长度 – 即连接队列最大长度.

当客户端发起了新的连接时, 回调函数需要为客户端套接字设置一个监视器, 并调用 uv_accept 函数将客户端套接字与新的监视器在关联一起. 在例子中我们将从流中读取数据.

void on_new_connection(uv_stream_t *server, int status) {if (status == -1) {// error!return;}uv_tcp_t *client = (uv_tcp_t*) malloc(sizeof(uv_tcp_t));uv_tcp_init(loop, client);if (uv_accept(server, (uv_stream_t*) client) == 0) {uv_read_start((uv_stream_t*) client, alloc_buffer, echo_read);}else {uv_close((uv_handle_t*) client, NULL);}
}

剩余部分的函数与上一节流式例子中的代码相似, 你可以在例子程序中找到具体代码, 如果套接字不再使用记得调用 uv_close 关闭该套接字. 如果你不再接受连接, 你可以在 uv_listen 的回调函数中关闭套接字.

客户端(Client)

在服务器端你需要调用 bind/listen/accept, 而在客户端你只需要调用 uv_tcp_connect.uv_tcp_connect 使用了与 uv_listen 风格相似的回调函数 uv_connect_cb 如下:

uv_tcp_t socket;
uv_tcp_init(loop, &socket);uv_connect_t connect;struct sockaddr_in dest = uv_ip4_addr("127.0.0.1", 80);uv_tcp_connect(&connect, &socket, dest, on_connect);

建立连接后会调用 on_connect.

UDP

User Datagram Protocol 提供了无连接, 不可靠网络通信协议, 因此 libuv 并不提供流式 UDP 服务, 而是通过uv_udp_t 结构体(用于接收)和 uv_udp_send_t 结构体(用于发送)以及相关的函数给开发人员提供了非阻塞的 UDP 服务. 所以, 真正读写 UDP 的函数与普通的流式读写非常相似.为了示范如何使用 UDP, 下面提供了一个简单的例子用来从DHCP 获取 IP 地址. – DHCP 发现.

uv_loop_t *loop;
uv_udp_t send_socket;
uv_udp_t recv_socket;int main() {loop = uv_default_loop();uv_udp_init(loop, &recv_socket);struct sockaddr_in recv_addr = uv_ip4_addr("0.0.0.0", 68);uv_udp_bind(&recv_socket, recv_addr, 0);uv_udp_recv_start(&recv_socket, alloc_buffer, on_read);uv_udp_init(loop, &send_socket);uv_udp_bind(&send_socket, uv_ip4_addr("0.0.0.0", 0), 0);uv_udp_set_broadcast(&send_socket, 1);uv_udp_send_t send_req;uv_buf_t discover_msg = make_discover_msg(&send_req);struct sockaddr_in send_addr = uv_ip4_addr("255.255.255.255", 67);uv_udp_send(&send_req, &send_socket, &discover_msg, 1, send_addr, on_send);return uv_run(loop, UV_RUN_DEFAULT);
}

首先我们在 68 号端口上设置了绑定本机所有网口的接收套接字(DHCP 客户端), 并且设置了读监视器. 然后我们利用相同的方法设置了一个用于发送消息的套接字. 并使用uv_udp_send 在 67 号端口上(DHCP 服务器)发送广播消息.

设置广播标志也是 必要 的, 不然你会得到 EACCES 错误[1]. 发送的具体消息与本书无关, 如果你对此感兴趣, 可以参考源码. 若出错, 则读写回调函数会收到 -1 状态码.

由于 UDP 套接字并不和特定的对等方保持连接, 所以 read 回调函数中将会收到用于标识发送者的额外信息. 如果缓冲区是由你自己的分配的, 并且不够容纳接收的数据, 则``flags`` 标志位可能是UV_UDP_PARTIAL. 在这种情况下, 操作系统会丢弃不能容纳的数据. (这也是 UDP 为你提供的特性).

void on_read(uv_udp_t *req, ssize_t nread, uv_buf_t buf, struct sockaddr *addr, unsigned flags) {if (nread == -1) {fprintf(stderr, "Read error %s\n", uv_err_name(uv_last_error(loop)));uv_close((uv_handle_t*) req, NULL);free(buf.base);return;}char sender[17] = { 0 };uv_ip4_name((struct sockaddr_in*) addr, sender, 16);fprintf(stderr, "Recv from %s\n", sender);// ... DHCP specific code
free(buf.base);uv_udp_recv_stop(req);
}

UV_EXTERN int uv_udp_set_membership(uv_udp_t* handle,const char* multicast_addr, const char* interface_addr,uv_membership membership);

DNS 查询(Querying DNS)

libuv 提供了异步解析 DNS 的功能, 用于替代 getaddrinfo[4]. 在回调函数中, 你可以在获得的 IP 地址上执行普通的套接字操作. 让我们通过一个简单的 DNS 解析的例子来看看怎么连接 Freenode 吧:

int main() {loop = uv_default_loop();struct addrinfo hints;hints.ai_family = PF_INET;hints.ai_socktype = SOCK_STREAM;hints.ai_protocol = IPPROTO_TCP;hints.ai_flags = 0;uv_getaddrinfo_t resolver;fprintf(stderr, "irc.freenode.net is... ");int r = uv_getaddrinfo(loop, &resolver, on_resolved, "irc.freenode.net", "6667", &hints);if (r) {fprintf(stderr, "getaddrinfo call error %s\n", uv_err_name(uv_last_error(loop)));return 1;}return uv_run(loop, UV_RUN_DEFAULT);
}

如果 uv_getaddrinfo 返回非零, 表示在建立连接时出错, 你设置的回调函数不会被调用, 所有的参数将会在uv_getaddrinfo 返回后被立即释放. 有关hostname, servname 和 hints 结构体的文档可以在 getaddrinfo 帮助页面中找到.

在解析回调函数中, 你可以在 struct addrinfo(s) 结构的链表中任取一个 IP. 这个例子也演示了如何使用 uv_tcp_connect. 你在回调函数中有必要调用 uv_freeaddrinfo.

void on_resolved(uv_getaddrinfo_t *resolver, int status, struct addrinfo *res) {if (status == -1) {fprintf(stderr, "getaddrinfo callback error %s\n", uv_err_name(uv_last_error(loop)));return;}char addr[17] = {'\0'};uv_ip4_name((struct sockaddr_in*) res->ai_addr, addr, 16);fprintf(stderr, "%s\n", addr);uv_connect_t *connect_req = (uv_connect_t*) malloc(sizeof(uv_connect_t));uv_tcp_t *socket = (uv_tcp_t*) malloc(sizeof(uv_tcp_t));uv_tcp_init(loop, socket);connect_req->data = (void*) socket;uv_tcp_connect(connect_req, socket, *(struct sockaddr_in*) res->ai_addr, on_connect);uv_freeaddrinfo(res);
}

#include <stdio.h>
#include <uv.h>int main() {char buf[512];uv_interface_address_t *info;int count, i;uv_interface_addresses(&info, &count);i = count;printf("Number of interfaces: %d\n", count);while (i--) {uv_interface_address_t interface = info[i];printf("Name: %s\n", interface.name);printf("Internal? %s\n", interface.is_internal ? "Yes" : "No");if (interface.address.address4.sin_family == AF_INET) {uv_ip4_name(&interface.address.address4, buf, sizeof(buf));printf("IPv4 address: %s\n", buf);}else if (interface.address.address4.sin_family == AF_INET6) {uv_ip6_name(&interface.address.address6, buf, sizeof(buf));printf("IPv6 address: %s\n", buf);}printf("\n");}uv_free_interface_addresses(info, count);return 0;
}