【libuv高效编程】libuv学习超详细教程11 —— libuv stream 流句柄解读
libuv 网络 Linux
网络
发布日期:   2020-04-24
作者:   杰杰
文章字数:   1.8k
阅读时长:   8 分
阅读次数:  

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;
}

参考

libuv官方文档

例程代码获取

libuv-learning-code

赏

  转载请注明: 杰杰的博客 【libuv高效编程】libuv学习超详细教程11 —— libuv stream 流句柄解读

 上一篇
闲谈：我为什么热爱开源。 闲谈：我为什么热爱开源。
hello！公众号的兄弟姐妹们大家好，我们今天来聊一聊我热爱的事情——开源。 回想几年前，我刚学C语言的时候，觉得这东西很好玩，也很有兴趣，居然能让我操控电脑，虽然能做的东西不对，也就打印点字符在屏幕上，但这确实却让我喜欢上了它，也许以前没
2020-05-17 闲谈开源
闲谈 开源 GitHub
下一篇 
【libuv高效编程】libuv学习超详细教程10 —— libuv stream 流句柄解读 【libuv高效编程】libuv学习超详细教程10 —— libuv stream 流句柄解读
libuv系列文章 【libuv高效编程】libuv学习超详细教程1——libuv的编译与安装 【libuv高效编程】libuv学习超详细教程2——libuv框架初窥 【libuv高效编程】libuv学习超详细教程3——libuv事件循
2020-04-24 网络
libuv 网络 Linux
  目录