libuv源码分析之stream第一篇

struct uv_stream_s {
  // uv_handle_s的字段
  void* data;        
  // 所属事件循环   
  uv_loop_t* loop;  
  // handle类型    
  uv_handle_type type;  
  // 关闭handle时的回调
  uv_close_cb close_cb; 
  // 用于插入事件循环的handle队列
  void* handle_queue[2];
  union {               
    int fd;             
    void* reserved[4];  
  } u;      
  // 用于插入事件循环的closing阶段对应的队列
  uv_handle_t* next_closing; 
  // 各种标记 
  unsigned int flags;
  // 流拓展的字段
  // 用户写入流的字节大小，流缓存用户的输入，然后等到可写的时候才做真正的写
  size_t write_queue_size; 
  // 分配内存的函数，内存由用户定义，主要用来保存读取的数据               
  uv_alloc_cb alloc_cb;  
  // 读取数据的回调                 
  uv_read_cb read_cb; 
  // 连接成功后，执行connect_req的回调（connect_req在uv__xxx_connect中赋值）
  uv_connect_t *connect_req; 
  // 关闭写端的时候，发送完缓存的数据，执行shutdown_req的回调（shutdown_req在uv_shutdown的时候赋值）    
  uv_shutdown_t *shutdown_req;
  // 流对应的io观察者，即文件描述符+一个文件描述符事件触发时执行的回调   
  uv__io_t io_watcher;  
  // 流缓存下来的，待写的数据         
  void* write_queue[2];       
  // 已经完成了数据写入的队列   
  void* write_completed_queue[2];
  // 完成三次握手后，执行的回调
  uv_connection_cb connection_cb;
  // 操作流时出错码
  int delayed_error;  
  // accept返回的通信socket对应的文件描述符           
  int accepted_fd;    
  // 同上，用于缓存更多的通信socket对应的文件描述符           
  void* queued_fds;
}

// 初始化流
void uv__stream_init(uv_loop_t* loop,
                     uv_stream_t* stream,
                     uv_handle_type type) {
  int err;
  // 记录handle的类型
  uv__handle_init(loop, (uv_handle_t*)stream, type);
  stream->read_cb = NULL;
  stream->alloc_cb = NULL;
  stream->close_cb = NULL;
  stream->connection_cb = NULL;
  stream->connect_req = NULL;
  stream->shutdown_req = NULL;
  stream->accepted_fd = -1;
  stream->queued_fds = NULL;
  stream->delayed_error = 0;
  QUEUE_INIT(&stream->write_queue);
  QUEUE_INIT(&stream->write_completed_queue);
  stream->write_queue_size = 0;
  // 这个逻辑看起来是为了拿到一个备用的文件描述符，如果以后触发UV_EMFILE错误（打开的文件太多）时，使用这个备用的fd
  if (loop->emfile_fd == -1) {
    err = uv__open_cloexec("/dev/null", O_RDONLY);
    if (err < 0)
        err = uv__open_cloexec("/", O_RDONLY);
    if (err >= 0)
      loop->emfile_fd = err;
  }
  // 初始化io观察者，把文件描述符（这里还没有，所以是-1）和回调uv__stream_io记录在io_watcher上
  uv__io_init(&stream->io_watcher, uv__stream_io, -1);
}

// 关闭nagle，开启长连接，保存fd 
int uv__stream_open(uv_stream_t* stream, int fd, int flags) {

  // 还没有设置fd或者设置的同一个fd则继续，否则返回busy
  if (!(stream->io_watcher.fd == -1 || stream->io_watcher.fd == fd))
    return UV_EBUSY;

  // 设置流的标记
  stream->flags |= flags;

  if (stream->type == UV_TCP) {
    // 关闭nagle算法
    if ((stream->flags & UV_HANDLE_TCP_NODELAY) && uv__tcp_nodelay(fd, 1))
      return UV__ERR(errno);

    // 开启SO_KEEPALIVE，使用tcp长连接，一定时间后没有收到数据包会发送心跳包
    if ((stream->flags & UV_HANDLE_TCP_KEEPALIVE) &&
        uv__tcp_keepalive(fd, 1, 60)) {
      return UV__ERR(errno);
    }
  }
   // 保存socket对应的文件描述符到io观察者中，libuv会在io poll阶段监听该文件描述符
  stream->io_watcher.fd = fd;

  return 0;
}

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 -ENOTCONN;
  // 标记正在读
  stream->flags |= UV_HANDLE_READING;
  // 记录读回调，有数据的时候会执行这个回调
  stream->read_cb = read_cb;
  // 分配内存函数，用于存储读取的数据
  stream->alloc_cb = alloc_cb;
  // 注册读事件
  uv__io_start(stream->loop, &stream->io_watcher, POLLIN);
  // 激活handle，有激活的handle，事件循环不会退出
  uv__handle_start(stream);

  return 0;
}

int uv_read_stop(uv_stream_t* stream) {
  // 是否正在执行读取操作，如果不是，则没有必要停止
  if (!(stream->flags & UV_HANDLE_READING))
    return 0;
  // 清除 正在读取 的标记
  stream->flags &= ~UV_HANDLE_READING;
  // 撤销 等待读事件
  uv__io_stop(stream->loop, &stream->io_watcher, POLLIN);
  // 对写事件也不感兴趣，停掉handle。允许事件循环退出
  if (!uv__io_active(&stream->io_watcher, POLLOUT))
    uv__handle_stop(stream);
  stream->read_cb = NULL;
  stream->alloc_cb = NULL;
  return 0;
}

int uv_is_readable(const uv_stream_t* stream) {
  return !!(stream->flags & UV_HANDLE_READABLE);
}

int uv_write(
       // 一个写请求，记录了需要写入的数据和信息。数据来自下面的const uv_buf_t bufs[]
         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);
}

int uv_write2(
    uv_write_t* req,
    uv_stream_t* stream,
    const uv_buf_t bufs[],
    unsigned int nbufs,
    // 需要传递的文件描述符所在的流，这里不分析，在分析unix的时候再分析
    uv_stream_t* send_handle,
    uv_write_cb cb
) 
{
  int empty_queue;
  // 是不可写流
  if (!(stream->flags & UV_HANDLE_WRITABLE))
    return -EPIPE;
  // 流中缓存的数据大小是否为0
  empty_queue = (stream->write_queue_size == 0);

  // 初始化一个写请求
  uv__req_init(stream->loop, req, UV_WRITE);
  // 写完后执行的回调
  req->cb = cb;
  // 往哪个流写
  req->handle = stream;
  // 写出错的错误码，初始化为0
  req->error = 0;
  QUEUE_INIT(&req->queue);
  // 默认buf
  req->bufs = req->bufsml;
  // 不够则扩容
  if (nbufs > ARRAY_SIZE(req->bufsml))
    req->bufs = uv__malloc(nbufs * sizeof(bufs[0]));
  // 把需要写入的数据填充到req中
  memcpy(req->bufs, bufs, nbufs * sizeof(bufs[0]));
  // 需要写入的buf个数
  req->nbufs = nbufs;
  // 目前写入的buf个数，初始化是0
  req->write_index = 0;
  // 更新流中待写数据的总长度，就是每个buf的数据大小加起来
  stream->write_queue_size += uv__count_bufs(bufs, nbufs);

  // 插入待写队列
  QUEUE_INSERT_TAIL(&stream->write_queue, &req->queue);
  /*
   stream->connect_req非空说明是作为客户端，并且正在建立三次握手，建立成功会置connect_req为NULL。
   这里非空说明还没有建立成功或者不是作为客户端（不是连接流）。即没有用到connect_req这个字段。
  */
  if (stream->connect_req) {
    /* Still connecting, do nothing. */
  }
  else if (empty_queue) { 
    // 待写队列为空，则直接触发写动作，即操作文件描述符
    uv__write(stream);
  }
  else {
    /*
        队列非空，说明往底层写，uv__write中不一样会注册等待可写事件，所以这里注册一下
        给流注册等待可写事件，触发的时候，把数据消费掉
    */
    uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
  }

  return 0;
}

static void uv__write(uv_stream_t* stream) {
  struct iovec* iov;
  QUEUE* q;
  uv_write_t* req;
  int iovmax;
  int iovcnt;
  ssize_t n;
  int err;

start:
  // 待写队列为空，没得写
  if (QUEUE_EMPTY(&stream->write_queue))
    return;
  // 遍历待写队列，把每个节点的数据写入底层
  q = QUEUE_HEAD(&stream->write_queue);
  req = QUEUE_DATA(q, uv_write_t, queue);
  /*
    struct iovec {
        ptr_t iov_base; // 数据首地址
        size_t iov_len; // 数据长度
    };  
    iovec和bufs结构体的定义一样
  */
  // 转成iovec格式发送
  iov = (struct iovec*) &(req->bufs[req->write_index]);
  // 待写的buf个数，nbufs是总数，write_index是当前已写的个数
  iovcnt = req->nbufs - req->write_index;
  // 最多能写几个
  iovmax = uv__getiovmax();

  // 取最小值
  if (iovcnt > iovmax)
    iovcnt = iovmax;

  // 有需要传递的描述符
  if (req->send_handle) {
    // 需要传递文件描述符的逻辑，分析unix域的时候再分析
  } else { // 单纯发送数据，则直接写
    do {
      if (iovcnt == 1) {
        n = write(uv__stream_fd(stream), iov[0].iov_base, iov[0].iov_len);
      } else {
        n = writev(uv__stream_fd(stream), iov, iovcnt);
      }
    } while (n == -1 && errno == EINTR);
  }
  // 发送出错
  if (n < 0) {
    // 发送失败的逻辑，我们不具体分析
  } else {
    // 写成功，n是写成功的字节数
    while (n >= 0) {
      // 本次待写数据的首地址
      uv_buf_t* buf = &(req->bufs[req->write_index]);
      // 某个buf的数据长度
      size_t len = buf->len;
      // len如果大于n说明本buf的数据部分被写入
      if ((size_t)n < len) {
        // 更新指针，指向下次待发送的数据首地址
        buf->base += n;
        // 更新待发送数据的长度
        buf->len -= n;
        // 更新待写数据的总长度
        stream->write_queue_size -= n;
        n = 0;
        // 设置了一直写标记，则继续写
        if (stream->flags & UV_HANDLE_BLOCKING_WRITES) {
          goto start;
        } else {
          // 否则等待可写事件触发的时候再写
          break;
        }
      } else {
        // 本buf的数据完成被写入，更新下一个待写入的buf位置
        req->write_index++;
        n -= len;
        // 更新待写数据总长度
        stream->write_queue_size -= len;
        // 如果写完了全部buf，触发回调
        if (req->write_index == req->nbufs) {
          // 写完了本请求的数据，做后续处理
          uv__write_req_finish(req);
          return;
        }
      }
    }
  }
  // 到这说明数据还没有完全被写入，注册等待可写事件，等待继续写
  uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);
  return;

error:
  // 写出错
  req->error = err;
  uv__write_req_finish(req);
  // 撤销等待可写事件
  uv__io_stop(stream->loop, &stream->io_watcher, POLLOUT);
  // 没有注册了等待可读事件，则停掉流
  if (!uv__io_active(&stream->io_watcher, POLLIN))
    uv__handle_stop(stream);
}

// 把buf的数据写入完成或写出错后触发的回调
static void uv__write_req_finish(uv_write_t* req) {
  uv_stream_t* stream = req->handle;
  // 移出队列
  QUEUE_REMOVE(&req->queue);
  // 写入成功了
  if (req->error == 0) {
    /*
      bufsml是默认的buf数，如果不够，则bufs指向新的内存，
      然后再储存数据。两者不等说明申请了额外的内存，需要free掉
    */ 
    if (req->bufs != req->bufsml)
      uv__free(req->bufs);
    req->bufs = NULL;
  }
  // 插入写完成队列
  QUEUE_INSERT_TAIL(&stream->write_completed_queue, &req->queue);
  // 插入pending队列，在pending阶段执行回调
  uv__io_feed(stream->loop, &stream->io_watcher);
}