libuv系列文章
stream handle 的外部API
我们接着上一篇文章接着讲解流操作的API。这些api接口我们会很经常使用到,比如在读取tcp、udp、文件数据的时候,就会用到,同理写数据的时候也会用到。
uv_shutdown()
关闭流的写端口,它会等待未完成的写操作,在关闭后通过uv_shutdown_cb指定的回调函数告知应用层。
注意了,它并不是关闭stream handle,只是关闭了写入端。
参数:
- req:指定关闭的请求。
- stream:指定stream handle。
- cb:在关闭后告知应用层的回调函数。
int uv_shutdown(uv_shutdown_t* req, uv_stream_t* stream, uv_shutdown_cb cb) {
/* 校验相关信息,只有 UV_TCP UV_TTY UV_NAMED_PIPE 类型的stream handle 才可以用*/
assert(stream->type == UV_TCP ||
stream->type == UV_TTY ||
stream->type == UV_NAMED_PIPE);
if (!(stream->flags & UV_HANDLE_WRITABLE) ||
stream->flags & UV_HANDLE_SHUT ||
stream->flags & UV_HANDLE_SHUTTING ||
uv__is_closing(stream)) {
return UV_ENOTCONN;
}
assert(uv__stream_fd(stream) >= 0);
/* 初始化请求 */
uv__req_init(stream->loop, req, UV_SHUTDOWN);
req->handle = stream;
req->cb = cb;
stream->shutdown_req = req;
/* 设置关闭标志位 */
stream->flags |= UV_HANDLE_SHUTTING;
/* 初始化io观察者,将 io 观察者加入到队列中后,以便在事件循环的特定阶段进行处理 */
uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
uv__stream_osx_interrupt_select(stream);
return 0;
}
uv_listen()
开始侦听新来的连接,如果你学习过TCP协议,那么对listen应该很熟悉,它就是用于监听连接的请求的。
函数实现:
int uv_listen(uv_stream_t* stream, int backlog, uv_connection_cb cb) {
int err;
/* 根据类型处理,只有UV_TCP UV_NAMED_PIPE 才可以 */
switch (stream->type) {
case UV_TCP:
/* tcp连接,调用tcp的listen去处理 */
err = uv_tcp_listen((uv_tcp_t*)stream, backlog, cb);
break;
case UV_NAMED_PIPE:
/* pipe连接,调用pipe的listen去处理 */
err = uv_pipe_listen((uv_pipe_t*)stream, backlog, cb);
break;
default:
err = UV_EINVAL;
}
if (err == 0)
uv__handle_start(stream);
return err;
}
参数:
stream:指定stream handle。
backlog:指定libuv监听的最大的连接数。
cb:连接的回调函数,当接受到新来的连接时,调用 uv_connection_cb 回调函数。
注意,这个函数只有tcp或者pipe类型的stream handle才可使用。
uv_accept()
调用用来配合 uv_listen() 接受新来的连接。一般来说会在 uv_connection_cb 的回调函数中去调用这个 uv_accept() 函数以接受连接,这与tcp协议的处理是非常像的。
注意:在调用这个函数前,客户端句柄必须被初始化。
int uv_accept(uv_stream_t* server, uv_stream_t* client) {
int err;
assert(server->loop == client->loop);
if (server->accepted_fd == -1)
return UV_EAGAIN;
switch (client->type) {
case UV_NAMED_PIPE:
case UV_TCP:
/* 获取文件描述符,并赋值给 client->io_watcher.fd */
err = uv__stream_open(client,
server->accepted_fd,
UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
if (err) {
/* 出现错误就关闭 */
uv__close(server->accepted_fd);
goto done;
}
break;
case UV_UDP:
/* 对于udp协议,通过uv_udp_open去获取文件描述符 */
err = uv_udp_open((uv_udp_t*) client, server->accepted_fd);
if (err) {
uv__close(server->accepted_fd);
goto done;
}
break;
default:
return UV_EINVAL;
}
client->flags |= UV_HANDLE_BOUND;
done:
/* 处理在排队的连接请求 */
if (server->queued_fds != NULL) {
uv__stream_queued_fds_t* queued_fds;
queued_fds = server->queued_fds;
/* 处理第一个排队的 */
server->accepted_fd = queued_fds->fds[0];
/* All read, free */
assert(queued_fds->offset > 0);
if (--queued_fds->offset == 0) {
uv__free(queued_fds);
server->queued_fds = NULL;
} else {
/* Shift rest */
memmove(queued_fds->fds,
queued_fds->fds + 1,
queued_fds->offset * sizeof(*queued_fds->fds));
}
} else {
server->accepted_fd = -1;
if (err == 0)
uv__io_start(server->loop, &server->io_watcher, POLLIN);
}
return err;
}
uv_read_start()
当连接成功后,可以调用uv_read_start()
函数去监听流的读取端,当有数据可读的时候,将会调用uv_read_cb
指定的回调函数,递交到用户去处理这些数据。
- 函数原型
int uv_read_start(uv_stream_t* stream,
uv_alloc_cb alloc_cb,
uv_read_cb read_cb)
参数:
stream:指定stream handle。
alloc_cb:读取数据时调用该函数分配内存空间。
read_cb:读取成功后触发异步回调。
函数源码:
int uv_read_start(uv_stream_t* stream,
uv_alloc_cb alloc_cb,
uv_read_cb read_cb) {
assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE ||
stream->type == UV_TTY);
if (stream->flags & UV_HANDLE_CLOSING)
return UV_EINVAL;
if (!(stream->flags & UV_HANDLE_READABLE))
return UV_ENOTCONN;
/* 设置标志位,表示此时流正在被使用读取 */
stream->flags |= UV_HANDLE_READING;
assert(uv__stream_fd(stream) >= 0);
assert(alloc_cb);
/* 注册回调函数 */
stream->read_cb = read_cb;
stream->alloc_cb = alloc_cb;
/* 启动io观察者 */
uv__io_start(stream->loop, &stream->io_watcher, POLLIN);
uv__handle_start(stream);
uv__stream_osx_interrupt_select(stream);
return 0;
}
uv_read_stop()
与uv_read_start()
函数刚好相反,uv_read_stop()
函数是停止从流读取数据。 uv_read_cb
回调函数将不再被调用。
参数:
- stream:指定stream handle。
int uv_read_stop(uv_stream_t* stream) {
/* 如果它未被设置为 UV_HANDLE_READING 读取占用标志位,则不需要停止 */
if (!(stream->flags & UV_HANDLE_READING))
return 0;
/* 取消设置 UV_HANDLE_READING 标志位 */
stream->flags &= ~UV_HANDLE_READING;
/* 停止流读取,关闭流io观察者 */
uv__io_stop(stream->loop, &stream->io_watcher, POLLIN);
/* 当且仅当io观察者不处于活跃时才关闭这个 stream handle */
if (!uv__io_active(&stream->io_watcher, POLLOUT))
uv__handle_stop(stream);
uv__stream_osx_interrupt_select(stream);
/* 取消注册回调函数 */
stream->read_cb = NULL;
stream->alloc_cb = NULL;
return 0;
}
uv_write()
向stream handle 写入数据,实际上是调用uv_write2()
这个函数。
参数:
req:请求。
handle:指定的stream handle
bufs:要写入的buf数据。
nbufs:要写入数据的大小。
cb:当写操作完成后,调用的回调函数。
int uv_write(uv_write_t* req,
uv_stream_t* handle,
const uv_buf_t bufs[],
unsigned int nbufs,
uv_write_cb cb) {
return uv_write2(req, handle, bufs, nbufs, NULL, cb);
}
uv_write2()函数
扩展的写函数,可用于在管道上发送数据。
int uv_write2(uv_write_t* req,
uv_stream_t* stream,
const uv_buf_t bufs[],
unsigned int nbufs,
uv_stream_t* send_handle,
uv_write_cb cb) {
int empty_queue;
/* 健壮性的判断,只有tcp pipe tty类型的handle才可用改函数 */
assert(nbufs > 0);
assert((stream->type == UV_TCP ||
stream->type == UV_NAMED_PIPE ||
stream->type == UV_TTY) &&
"uv_write (unix) does not yet support other types of streams");
if (uv__stream_fd(stream) < 0)
return UV_EBADF;
if (!(stream->flags & UV_HANDLE_WRITABLE))
return UV_EPIPE;
if (send_handle) {
if (stream->type != UV_NAMED_PIPE || !((uv_pipe_t*)stream)->ipc)
return UV_EINVAL;
if (uv__handle_fd((uv_handle_t*) send_handle) < 0)
return UV_EBADF;
#if defined(__CYGWIN__) || defined(__MSYS__)
/* Cygwin recvmsg always sets msg_controllen to zero, so we cannot send it.
See https://github.com/mirror/newlib-cygwin/blob/86fc4bf0/winsup/cygwin/fhandler_socket.cc#L1736-L1743 */
return UV_ENOSYS;
#endif
}
/* 即使write_queue为空,write_queue_size> 0也合法;这意味着write_completed_queue中存在错误状态请求,这些错误状态请求将在以后修改write_queue_size */
empty_queue = (stream->write_queue_size == 0);
/* 初始化请求 */
uv__req_init(stream->loop, req, UV_WRITE);
/* 注册回调函数,注册请求的stream handle */
req->cb = cb;
req->handle = stream;
req->error = 0;
req->send_handle = send_handle;
QUEUE_INIT(&req->queue);
req->bufs = req->bufsml;
if (nbufs > ARRAY_SIZE(req->bufsml))
req->bufs = uv__malloc(nbufs * sizeof(bufs[0]));
if (req->bufs == NULL)
return UV_ENOMEM;
memcpy(req->bufs, bufs, nbufs * sizeof(bufs[0]));
req->nbufs = nbufs;
req->write_index = 0;
stream->write_queue_size += uv__count_bufs(bufs, nbufs);
/* 将请求追加到write_queue。 */
QUEUE_INSERT_TAIL(&stream->write_queue, &req->queue);
/* 如果此函数开始时队列为空,则应尝试立即进行写操作。否则,启动write_watcher并等待fd变为可写状态。 */
if (stream->connect_req) {
/* 仍在连接,什么也不做. */
}
else if (empty_queue) {
uv__write(stream);
}
else {
/* 阻塞流永远不会在队列中有任何东西。 */
assert(!(stream->flags & UV_HANDLE_BLOCKING_WRITES));
uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
uv__stream_osx_interrupt_select(stream);
}
return 0;
}