node.js的http.createServer过程深入解析

var http = require('http');
http.createserver(function (request, response) {
  response.end('hello world
');
}).listen(9297);

function createserver(requestlistener) {
 return new server(requestlistener);
}

function server(requestlistener) {
 if (!(this instanceof server)) return new server(requestlistener);
 net.server.call(this, { allowhalfopen: true });
 // 收到http请求时执行的回调
 if (requestlistener) {
  this.on('request', requestlistener);
 }

 this.httpallowhalfopen = false;
 // 建立tcp连接的回调
 this.on('connection', connectionlistener);

 this.timeout = 2 * 60 * 1000;
 this.keepalivetimeout = 5000;
 this._pendingresponsedata = 0;
 this.maxheaderscount = null;
}
util.inherits(server, net.server);

function server(options, connectionlistener) {
 if (!(this instanceof server))
  return new server(options, connectionlistener);

 eventemitter.call(this);
 // connectionlistener在http.js处理过了
 if (typeof options === 'function') {
  connectionlistener = options;
  options = {};
  this.on('connection', connectionlistener);
 } else if (options == null || typeof options === 'object') {
  options = options || {};

  if (typeof connectionlistener === 'function') {
   this.on('connection', connectionlistener);
  }
 } else {
  throw new errors.typeerror('err_invalid_arg_type',
                'options',
                'object',
                options);
 }

 this._connections = 0;
 ......
 this[async_id_symbol] = -1;
 this._handle = null;
 this._usingworkers = false;
 this._workers = [];
 this._unref = false;

 this.allowhalfopen = options.allowhalfopen || false;
 this.pauseonconnect = !!options.pauseonconnect;
}

server.prototype.listen = function(...args) {
 // 处理入参，根据文档我们知道listen可以接收好几个参数，我们这里是只传了端口号9297
 var normalized = normalizeargs(args);
 // normalized = [{port: 9297}, null];
 var options = normalized[0];
 var cb = normalized[1];
 // 第一次listen的时候会创建，如果非空说明已经listen过
 if (this._handle) {
  throw new errors.error('err_server_already_listen');
 }
 ......
 listenincluster(this, null, options.port | 0, 4,
           backlog, undefined, options.exclusive);
}
function listenincluster() {
  ...
  server._listen2(address, port, addresstype, backlog, fd);
}

_listen2 = setuplistenhandle = function() {
  ......
  this._handle = createserverhandle(...);
  this._handle.listen(backlog || 511);
}
function createserverhandle() {
  handle = new tcp(tcpconstants.server);
  handle.bind(address, port);
}

void tcpwrap::new(const functioncallbackinfo<value>& args) {
 // this constructor should not be exposed to public javascript.
 // therefore we assert that we are not trying to call this as a
 // normal function.
 check(args.isconstructcall());
 check(args[0]->isint32());
 environment* env = environment::getcurrent(args);

 int type_value = args[0].as<int32>()->value();
 tcpwrap::sockettype type = static_cast<tcpwrap::sockettype>(type_value);

 providertype provider;
 switch (type) {
  case socket:
   provider = provider_tcpwrap;
   break;
  case server:
   provider = provider_tcpserverwrap;
   break;
  default:
   unreachable();
 }

 new tcpwrap(env, args.this(), provider);
}


tcpwrap::tcpwrap(environment* env, local<object> object, providertype provider)
  : connectionwrap(env, object, provider) {
 int r = uv_tcp_init(env->event_loop(), &handle_);
 check_eq(r, 0); 
}

uv_tcp_t
uv_tcp_t
// 初始化一个tcp流的结构体
int uv_tcp_init(uv_loop_t* loop, uv_tcp_t* tcp) {
 // 未指定未指定协议
 return uv_tcp_init_ex(loop, tcp, af_unspec);
}

int uv_tcp_init_ex(uv_loop_t* loop, uv_tcp_t* tcp, unsigned int flags) {
 int domain;

 /* use the lower 8 bits for the domain */
 // 低八位是domain
 domain = flags & 0xff;
 if (domain != af_inet && domain != af_inet6 && domain != af_unspec)
  return uv_einval;
 // 除了第八位的其他位是flags
 if (flags & ~0xff)
  return uv_einval;

 uv__stream_init(loop, (uv_stream_t*)tcp, uv_tcp);

 /* if anything fails beyond this point we need to remove the handle from
  * the handle queue, since it was added by uv__handle_init in uv_stream_init.
  */

 if (domain != af_unspec) {
  int err = maybe_new_socket(tcp, domain, 0);
  if (err) {
   // 出错则把该handle移除loop队列
   queue_remove(&tcp->handle_queue);
   return err;
  }
 }

 return 0;
}

void uv__stream_init(uv_loop_t* loop,
           uv_stream_t* stream,
           uv_handle_type type) {
 int err;

 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;

 if (loop->emfile_fd == -1) {
  err = uv__open_cloexec("/dev/null", o_rdonly);
  if (err < 0)
    /* in the rare case that "/dev/null" isn't mounted open "/"
     * instead.
     */
    err = uv__open_cloexec("/", o_rdonly);
  if (err >= 0)
   loop->emfile_fd = err;
 }

#if defined(__apple__)
 stream->select = null;
#endif /* defined(__apple_) */
 // 初始化io观察者
 uv__io_init(&stream->io_watcher, uv__stream_io, -1);
}

void uv__io_init(uv__io_t* w, uv__io_cb cb, int fd) {
 assert(cb != null);
 assert(fd >= -1);
 // 初始化队列，回调，需要监听的fd
 queue_init(&w->pending_queue);
 queue_init(&w->watcher_queue);
 w->cb = cb;
 w->fd = fd;
 w->events = 0;
 w->pevents = 0;

#if defined(uv_have_kqueue)
 w->rcount = 0;
 w->wcount = 0;
#endif /* defined(uv_have_kqueue) */
}

/* cannot set ipv6-only mode on non-ipv6 socket. */
 if ((flags & uv_tcp_ipv6only) && addr->sa_family != af_inet6)
  return uv_einval;
 // 获取一个socket并且设置某些标记
 err = maybe_new_socket(tcp, addr->sa_family, 0);
 if (err)
  return err;

 on = 1;
 // 设置在端口可重用
 if (setsockopt(tcp->io_watcher.fd, sol_socket, so_reuseaddr, &on, sizeof(on)))
  return uv__err(errno);
 bind(tcp->io_watcher.fd, addr, addrlen) && errno != eaddrinuse
static int maybe_new_socket(uv_tcp_t* handle, int domain, unsigned long flags) {
 struct sockaddr_storage saddr;
 socklen_t slen;

 if (domain == af_unspec) {
  handle->flags |= flags;
  return 0;
 }
 return new_socket(handle, domain, flags);
}
static int new_socket(uv_tcp_t* handle, int domain, unsigned long flags) {
 struct sockaddr_storage saddr;
 socklen_t slen;
 int sockfd;
 int err;
 // 获取一个socket
 err = uv__socket(domain, sock_stream, 0);
 if (err < 0)
  return err;
 sockfd = err;
 // 设置选项和保存socket的文件描述符到io观察者中
 err = uv__stream_open((uv_stream_t*) handle, sockfd, flags);
 if (err) {
  uv__close(sockfd);
  return err;
 }
 ...
 return 0;
}

int uv__stream_open(uv_stream_t* stream, int fd, int flags) {
 if (!(stream->io_watcher.fd == -1 || stream->io_watcher.fd == fd))
  return uv_ebusy;

 assert(fd >= 0);
 stream->flags |= flags;

 if (stream->type == uv_tcp) {
  if ((stream->flags & uv_handle_tcp_nodelay) && uv__tcp_nodelay(fd, 1))
   return uv__err(errno);

  /* todo use delay the user passed in. */
  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;
}

void tcpwrap::listen(const functioncallbackinfo<value>& args) {
 tcpwrap* wrap;
 assign_or_return_unwrap(&wrap,
             args.holder(),
             args.getreturnvalue().set(uv_ebadf));
 int backlog = args[0]->int32value();
 int err = uv_listen(reinterpret_cast<uv_stream_t*>(&wrap->handle_),
           backlog,
           onconnection);
 args.getreturnvalue().set(err);
}

 if (listen(tcp->io_watcher.fd, backlog))
  return uv__err(errno);
 // cb即onconnection
 tcp->connection_cb = cb;
 tcp->flags |= uv_handle_bound;

 // 有连接到来时的libuv层回调，覆盖了uv_stream_init时设置的值
 tcp->io_watcher.cb = uv__server_io;
 // 注册事件
 uv__io_start(tcp->loop, &tcp->io_watcher, pollin);

// 继续注册事件，等待连接
 uv__io_start(stream->loop, &stream->io_watcher, pollin);
 err = uv__accept(uv__stream_fd(stream));
 // 保存连接对应的socket
 stream->accepted_fd = err;
 // 执行nodejs层回调
 stream->connection_cb(stream, 0);

onconnection(uv_stream_t* handle,int status)
  if (status == 0) {
    // 新建一个uv_tcp_t结构体
    local<object> client_obj = wraptype::instantiate(env, wrap_data, wraptype::socket);
    wraptype* wrap;
    assign_or_return_unwrap(&wrap, client_obj);
    uv_stream_t* client_handle = reinterpret_cast<uv_stream_t*>(&wrap->handle_);
    // uv_accept返回0表示成功
    if (uv_accept(handle, client_handle))
     return;
    argv[1] = client_obj;
 }
 // 执行上层的回调，该回调是net.js设置的onconnection
 wrap_data->makecallback(env->onconnection_string(), arraysize(argv), argv);

int uv_accept(uv_stream_t* server, uv_stream_t* client) {
  ...
  // 新建的uv_tcp_t结构体关联accept_fd，注册读写事件
  uv__stream_open(client, server->accepted_fd, uv_handle_readable | uv_handle_writable);
  ...
}

function onconnection(err, clienthandle) {
 var handle = this;
 var self = handle.owner;
 if (err) {
  self.emit('error', errnoexception(err, 'accept'));
  return;
 }
 if (self.maxconnections && self._connections >= self.maxconnections) {
  clienthandle.close();
  return;
 }
 var socket = new socket({
  handle: clienthandle,
  allowhalfopen: self.allowhalfopen,
  pauseoncreate: self.pauseonconnect
 });
 socket.readable = socket.writable = true;
 self._connections++;
 socket.server = self;
 socket._server = self;
 dtrace_net_server_connection(socket);
 lttng_net_server_connection(socket);
 counter_net_server_connection(socket);
 // 触发_http_server.js里设置的connectionlistener回调
 self.emit('connection', socket);
}