// 創(chuàng)建一個(gè) socket，拿到一個(gè)文件描述符
int server_fd = socket();
// 綁定地址（IP + 端口）到該 socket 中
bind(server_fd, addressInfo);
// 修改 socket 為監(jiān)聽狀態(tài)，這樣就可以接收 TCP 連接了
listen(server_fd);

while(1) {
    int client_fd  = accept(server_fd);
    read(client_fd);
    write(client_fd);
}

while(1) {  
    int client_fd = accept(socket); 
    // 忽略出錯(cuò)處理 
    if (fork() > 0) {  
        continue;
        // 父進(jìn)程負(fù)責(zé) accept  
    } else {  
        // 子進(jìn)程負(fù)責(zé)處理連接
        handle(client_fd); 
        exit(); 
    }  
}

let fds = [[], [], [], …進(jìn)程個(gè)數(shù)];  
let process = [];  
for (let i = 0 ; i < 進(jìn)程個(gè)數(shù); i++) {  
    // 創(chuàng)建管道用于傳遞文件描述符  
    socketpair(fds[i]);  
    let pid;  
    if (pid = fork() > 0) {  
        // 父進(jìn)程  
        process.push({pid, 其它字段});  
    } else {  
        const index = i;  
        // 子進(jìn)程處理請(qǐng)求  
        while(1) {  
            // 從管道中讀取文件描述符  
            var client_fd = read(fd[index][1]);  
            // 處理請(qǐng)求  
            handle(client_fd);  
        }  
    }  
}  
// 主進(jìn)程 accept
for (;;) {  
    const clientFd = accept(socket);  
    // 找出處理該請(qǐng)求的子進(jìn)程  
    const i = findProcess();  
    // 傳遞文件描述符  
    write(fds[i][0], clientFd);  
}

int server_fd = socket();
bind(server_fd);
for (let i = 0 ; i < 進(jìn)程個(gè)數(shù); i++) {  
    if (fork() > 0) {  
        // 父進(jìn)程負(fù)責(zé)監(jiān)控子進(jìn)程  
    } else {  
        // 子進(jìn)程處理請(qǐng)求  
        listen(server_fd);
        while(1) {  
            int client_fd = accept(socket);  
            handle(client_fd);  
        }  
    }  
}

// 創(chuàng)建一個(gè) epoll 實(shí)例
int epoll_fd = epoll_create();  
/* 
    在 epoll 給某個(gè)文件描述符注冊(cè)感興趣的事件，這里是監(jiān)聽的 socket，注冊(cè)可讀事件，即連接到來 
    event = { 
        event: 可讀 
        fd：監(jiān)聽 socket 
        // 一些上下文 
    } 
*/  
epoll_ctl(epoll_fd , EPOLL_CTL_ADD , socket, event);  
while(1) {  
    // 阻塞等待事件就緒，events 保存就緒事件的信息，total 是個(gè)數(shù)  
    int total= epoll_wait(epoll_fd , 保存就緒事件的結(jié)構(gòu)events, 事件個(gè)數(shù), timeout);  
    for (let i = 0; i < total; i++) {  
        if (events[fd] === 監(jiān)聽 socket) {  
            int client_fd = accpet(socket);  
            // 把新的 socket 也注冊(cè)到 epoll，等待可讀，即可讀取客戶端數(shù)據(jù)  
            epoll_ctl(epoll_fd , EPOLL_CTL_ADD , client_fd, event);  
        } else {  
            //  從events[i] 中拿到一些上下文，執(zhí)行相應(yīng)的回調(diào)  
        }  
    }  
}

function epoll_wait() {  
    for 事件個(gè)數(shù)  
        // 調(diào)用文件系統(tǒng)的函數(shù)判斷  
        if (事件 [i] 中對(duì)應(yīng)的文件描述符中有某個(gè)用戶感興趣的事件發(fā)生 ？) {  
            插入就緒事件隊(duì)列  
        } else {  
            /*
                在事件 [i] 中的文件描述符所對(duì)應(yīng)的文件 / socke / 管道等資源中注冊(cè)回調(diào)。
                感興趣的事件觸發(fā)后回調(diào) epoll，回調(diào) epoll 后，epoll 把該 event[i] 插入
                就緒事件隊(duì)列返回給用戶  
            */
        }  
}

// 沒有空間可寫了
while(!(space = UN_BUF_SPACE(pupd))) 
{
    // 非阻塞模式，直接返回錯(cuò)誤碼
    if (nonblock) 
        return(-EAGAIN);
    // 阻塞模式，進(jìn)入阻塞狀態(tài)
    interruptible_sleep_on(sock->wait);
}

void interruptible_sleep_on(struct wait_queue **p)
{
    // 修改線程狀態(tài)為阻塞狀態(tài)
    __sleep_on(p,TASK_INTERRUPTIBLE);
}

static inline void __sleep_on(struct wait_queue **p, int state)
{
    unsigned long flags;
    // current 代表當(dāng)前執(zhí)行的線程
    struct wait_queue wait = { current, NULL };
    // 修改線程狀態(tài)為阻塞狀態(tài)
    current->state = state;
    // 當(dāng)前線程加入到資源的阻塞隊(duì)列，資源就緒后喚醒線程
    add_wait_queue(p, &wait);
    // 重新調(diào)度其他線程執(zhí)行，即從就緒的線程中選擇一個(gè)來執(zhí)行
    schedule();
}

#include <sys/event.h>
#include <fcntl.h>
#include <stdio.h>

int main(int argc, char **argv)
{   
    // 用于注冊(cè)事件到 kqueue
    struct  kevent event;
    // 用于接收從 kqueue 返回到事件，表示哪個(gè) fd 觸發(fā)了哪些事件
    struct  kevent emit_event;
    int kqueue_fd, file_fd, result;
    // 打開需要監(jiān)控的文件，拿到一個(gè) fd
    file_fd = open(argv[1], O_RDONLY);
    if (file_fd == -1) {
      printf("Fail to open %s", argv[1]);
      return 1;
    }
    // 創(chuàng)建 kqueue 實(shí)例
    kqueue_fd = kqueue();
    // 設(shè)置需要監(jiān)聽的事件，文件被寫入時(shí)觸發(fā)
    EV_SET(&event,file_fd, EVFILT_VNODE, EV_ADD | EV_CLEAR, NOTE_RENAME, 0, NULL);
    // 注冊(cè)到操作系統(tǒng)
    result = kevent(kqueue_fd, &event, 1, NULL, 0, NULL);
    // 不斷阻塞等待，直到文件被寫入
    while(1) {
        // result 返回觸發(fā)事件的 fd 個(gè)數(shù)，這里是一個(gè)
        result = kevent(kqueue_fd, NULL, 0, &emit_event,  1, NULL);
        if (result > 0) {
            printf("%s have been renamed\n", argv[1]);
        }
    }
}

uv_loop_t* loop;
napi_get_uv_event_loop(env, &loop);
struct io_uring_info *io_uring_data = (io_uring_info *)loop->data;
// 申請(qǐng)內(nèi)存
struct request *req = (struct request *)malloc(sizeof(*req) + (sizeof(struct iovec) * 1));
req->fd = fd;
req->offset = offset;
// 保存回調(diào)
napi_create_reference(env, args[2], 1, &req->func);
req->env = env;
req->nvecs = 1;
// 記錄buffer大小
req->iovecs[0].iov_len = bufferLength;
// 記錄內(nèi)存地址
req->iovecs[0].iov_base = bufferData;
// 提交給操作系統(tǒng)，操作系統(tǒng)讀完后通知線程，op 為 IORING_OP_READV 表示讀操作
submit_request(op, req, &io_uring_data->ring);

// 創(chuàng)建一個(gè) TCP Server
const server = net.createServer((socket) => {
  // 處理連接
});

// 監(jiān)聽端口，啟動(dòng)服務(wù)器
server.listen(8888);

function createServer(options, connectionListener) {  
  return new Server(options, connectionListener);  
}  

function Server(options, connectionListener) {  
  EventEmitter.call(this);  
  // 服務(wù)器收到的連接數(shù)，可以通過 maxConnections 限制并發(fā)連接數(shù)  
  this._connections = 0;  
  // C++ 層的對(duì)象，真正實(shí)現(xiàn) TCP 功能的地方
  this._handle = null;  
  // 服務(wù)器下的連接是否允許半關(guān)閉 
  this.allowHalfOpen = options.allowHalfOpen || false;  
  // 有連接時(shí)是否注冊(cè)可讀事件，如果該 socket 是交給其他進(jìn)程處理的話可以設(shè)置為 true 
  this.pauseOnConnect = !!options.pauseOnConnect;  
}

Server.prototype.listen = function(...args) {  
  /*
    處理入?yún)?，listen 可以接收很多種格式的參數(shù)，
    假設(shè)我們這里只傳了 8888 端口號(hào)
  */
    const normalized = normalizeArgs(args);  
    //  normalized = [{port: 8888}, null];  
    const options = normalized[0]; 
    // 監(jiān)聽成功后的回調(diào) 
    const cb = normalized[1];  

    // listen 成功后執(zhí)行的回調(diào)  
    if (cb !== null) {
        this.once('listening', cb);
    }
    listenIncluster(this, 
                    null, 
                    options.port | 0, 
                    4,      
                    ...); 
    return this;  
};

function listenIncluster(server, 
                         address, 
                         port, 
                         addressType,      
                         backlog, 
                         fd, 
                         exclusive) {  
  exclusive = !!exclusive;  
  if (cluster === null) cluster = require('cluster'); 
  if (cluster.isMaster || exclusive) {  
    server._listen2(address, port, addressType, backlog, fd);
    return;  
  }  
}

function setupListenHandle(address, port, addressType, backlog, fd) {  
    // 通過 C++ 層導(dǎo)出的 API 創(chuàng)建一個(gè)對(duì)象，該對(duì)象關(guān)聯(lián)了 C++ 層的 TCPWrap 對(duì)象
    this._handle = new TCP(TCPConstants.SERVER);
    // 創(chuàng)建 socket 并綁定地址到 socket 中
    this._handle.bind(address, port); 
    // 有完成三次握手的連接時(shí)執(zhí)行的回調(diào)  
    this._handle.onconnection = onconnection;  
    // 互相關(guān)聯(lián)
    this._handle.owner = this;  
    // 執(zhí)行 C++ 層 listen  
    this._handle.listen(backlog || 511);  
    // 觸發(fā) listen 回調(diào)  
    nextTick(this[async_id_symbol], emitListeningNT, this);  
}

void TCPWrap::New(const FunctionCallbackInfo<Value>& args) {
  new TCPWrap(env, args.This(), ...);
}

TCPWrap::TCPWrap(Environment* env, Local<Object> object, ProviderType provider)
    : ConnectionWrap(env, object, provider) {
  // 初始化一個(gè) tcp handle
  int r = uv_tcp_init(env->event_loop(), &handle_);
}

template <typename T>
void TCPWrap::Bind(...) {
  // 通過 JS 對(duì)象拿到關(guān)聯(lián)的 C++ TCPWrap 對(duì)象
  TCPWrap* wrap;
  ASSIGN_OR_RETURN_UNWRAP(&wrap,
                          args.Holder(),
                          args.GetReturnValue().Set(UV_EBADF));
  // 通過 JS 傳入的地址信息直接調(diào)用 Libuv
  uv_tcp_bind(&wrap->handle_,
              reinterpret_cast<const sockaddr*>(&addr),
              flags);
}

int uv_tcp_bind(...) {
  return uv__tcp_bind(handle, addr, addrlen, flags);
}

int uv__tcp_bind(uv_tcp_t* tcp,
                 const struct sockaddr* addr,
                 unsigned int addrlen,
                 unsigned int flags) {
  // 創(chuàng)建一個(gè) socket，并把返回的 fd 保存到 tcp 結(jié)構(gòu)體中
  maybe_new_socket(tcp, addr->sa_family, 0);

  on = 1;
  // 默認(rèn)設(shè)置了 SO_REUSEADDR 屬性，后面具體分析
  setsockopt(tcp->io_watcher.fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
  // 綁定地址信息到 socket
  bind(tcp->io_watcher.fd, addr, addrlen);
  return 0;
}

int uv_listen(uv_stream_t* stream, int backlog, uv_connection_cb cb) {
  uv_tcp_listen((uv_tcp_t*)stream, backlog, cb);
}

int uv_tcp_listen(uv_tcp_t* tcp, int backlog, uv_connection_cb cb) {
  static int single_accept = -1;
  unsigned long flags;
  int err;
  // 已廢棄
  if (single_accept == -1) {
    const char* val = getenv("UV_TCP_SINGLE_ACCEPT");
    single_accept = (val != NULL && atoi(val) != 0); 
  }
  // 有連接時(shí)是否連續(xù)接收，或者間歇性處理，見后面分析
  if (single_accept)
    tcp->flags |= UV_HANDLE_TCP_SINGLE_ACCEPT;

  flags = 0;
  // 設(shè)置 flags 到 handle 上，因?yàn)橐呀?jīng)創(chuàng)建了 socket
  maybe_new_socket(tcp, AF_INET, flags);
  listen(tcp->io_watcher.fd, backlog)
  // 保存回調(diào)，有連接到來時(shí)被 Libuv 執(zhí)行
  tcp->connection_cb = cb;
  tcp->flags |= UV_HANDLE_BOUND;
  // 有連接來時(shí)的處理函數(shù)，該函數(shù)再執(zhí)行上面的 connection_cb
  tcp->io_watcher.cb = uv__server_io;
  // 注冊(cè)可讀事件，等待連接到來
  uv__io_start(tcp->loop, &tcp->io_watcher, POLLIN);

  return 0;
}

void uv__server_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) {
  uv_stream_t* stream;
  int err;

  stream = container_of(w, uv_stream_t, io_watcher);

  uv__io_start(stream->loop, &stream->io_watcher, POLLIN);
  // 回調(diào)了可能關(guān)閉了 server，所以需要實(shí)時(shí)判斷
  while (uv__stream_fd(stream) != -1) {
    // 摘取一個(gè) TCP 連接，成功的話，err 保存了對(duì)應(yīng)的 fd
    err = uv__accept(uv__stream_fd(stream));
    // 保存 fd 在 accepted_fd，等待處理
    stream->accepted_fd = err;
    // 執(zhí)行回調(diào)
    stream->connection_cb(stream, 0);
    // 如果回調(diào)里沒有處理該 accepted_fd，則注銷可讀事件、先不處理新的連接
    if (stream->accepted_fd != -1) {
      uv__io_stop(loop, &stream->io_watcher, POLLIN);
      return;
    }
    // 設(shè)置了 UV_HANDLE_TCP_SINGLE_ACCEPT 則進(jìn)入睡眠，讓其他進(jìn)程有機(jī)會(huì)參與處理
    if (stream->type == UV_TCP &&
        (stream->flags & UV_HANDLE_TCP_SINGLE_ACCEPT)) {
      struct timespec timeout = { 0, 1 };
      nanosleep(&timeout, NULL);
    }
  }
}

// WrapType 為 TCPWrap，UVType 為 uv_tcp_t
template <typename WrapType, typename UVType>  
void ConnectionWrap<WrapType, UVType>::OnConnection(uv_stream_t* handle, int status) {  
    // HandleWrap 中保存了 handle 和 TCPWrap 的關(guān)系，這里取出來使用               
    WrapType* wrap_data = static_cast<WrapType*>(handle->data);  
    Environment* env = wrap_data->env();  
    Local<Value> argv[] = {  
        Integer::New(env->isolate(), status),  
        Undefined(env->isolate())  
    };  

    // 新建一個(gè)表示和客戶端通信的對(duì)象，和 JS 層執(zhí)行 new TCP 一樣 
    Local<Object> client_obj = WrapType::Instantiate(env,wrap_data,WrapType::SOCKET);  
    WrapType* wrap;  
    // 從 client_obj 中取出關(guān)聯(lián)的 TCPWrap 對(duì)象存到 wrap 中 
    ASSIGN_OR_RETURN_UNWRAP(&wrap, client_obj);  
    // 拿到 TCPWrap 中的 uv_tcp_t 結(jié)構(gòu)體，再轉(zhuǎn)成 uv_stream_t，因?yàn)樗鼈冾愃聘割惡妥宇惖年P(guān)系
    uv_stream_t* client_handle = reinterpret_cast<uv_stream_t*>(&wrap->handle_);  
    // 把通信 fd 存儲(chǔ)到 client_handle 中  
    uv_accept(handle, client_handle);
    argv[1] = client_obj;  
    // 回調(diào)上層的 onconnection 函數(shù)  
    wrap_data->MakeCallback(env->onconnection_string(), arraysize(argv), argv);  
}

MaybeLocal<Object> TCPWrap::Instantiate(Environment* env,
                                        AsyncWrap* parent,
                                        TCPWrap::SocketType type) {
  // 拿到導(dǎo)出到 JS 層的 TCP 構(gòu)造函數(shù)（緩存在env中）
  Local<Function> constructor = env->tcp_constructor_template()
                                    ->GetFunction(env->context())
                                    .ToLocalChecked();
  Local<Value> type_value = Int32::New(env->isolate(), type);
  // 相當(dāng)于我們?cè)?JS 層調(diào)用 new TCP() 時(shí)拿到的對(duì)象
  return handle_scope.EscapeMaybe(
      constructor->NewInstance(env->context(), 1, &type_value));
}

int uv_accept(uv_stream_t* server, uv_stream_t* client) {
  int err;
  // 把 accepted_fd 保存到 client 中
  uv__stream_open(client,
                  server->accepted_fd,
                  UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
  // 處理了，重置該字段
  server->accepted_fd = -1;
  // 保證注冊(cè)了可讀事件，繼續(xù)處理新的連接
  uv__io_start(server->loop, &server->io_watcher, POLLIN);
  return err;
}

// clientHandle 代表一個(gè)和客戶端建立 TCP 連接的實(shí)體  
function onconnection(err, clientHandle) {  
  const handle = this;  
  const self = handle.owner;  
  // 新建一個(gè) socket 用于通信  
  const socket = new Socket({  
    handle: clientHandle,  
    allowHalfOpen: self.allowHalfOpen,  
    pauseOnCreate: self.pauseOnConnect  
  });  
  // 服務(wù)器的連接數(shù)加一  
  self._connections++;  
  // 觸發(fā)用戶層連接事件  
  self.emit('connection', socket); 
}

const http = require('http');  
http.createServer((req, res) => {  
  res.write('hello');  
  res.end();  
})  
.listen(3000);

function createServer(opts, requestListener) {  
  return new Server(opts, requestListener);  
}

function Server(options, requestListener) {  
  // 可以自定義表示請(qǐng)求的對(duì)象和響應(yīng)的對(duì)象  
  this[kIncomingMessage] = options.IncomingMessage || IncomingMessage;  
  this[kServerResponse] = options.ServerResponse || ServerResponse;  
  // HTTP 頭最大字節(jié)數(shù)  
  this.maxHeaderSize = options.maxHeaderSize;  
  // 允許半關(guān)閉  
  net.Server.call(this, { allowHalfOpen: true });  
  // 有請(qǐng)求時(shí)的回調(diào)  
  if (requestListener) {  
    this.on('request', requestListener);  
  }  
  // 服務(wù)器 socket 讀端關(guān)閉時(shí)是否允許繼續(xù)處理隊(duì)列里的響應(yīng)（TCP 上有多個(gè)請(qǐng)求，管道化）  
  this.httpAllowHalfOpen = false;  
  // 有連接時(shí)的回調(diào)，由 net 模塊觸發(fā)  
  this.on('connection', connectionListener);  
  // 服務(wù)器下所有請(qǐng)求和響應(yīng)的超時(shí)時(shí)間  
  this.timeout = 0;  
  // 同一個(gè) TCP 連接上，兩個(gè)請(qǐng)求之前最多間隔的時(shí)間   
  this.keepAliveTimeout = 5000;  
  // HTTP 頭的最大個(gè)數(shù)
  this.maxHeadersCount = null;  
  // 解析頭部的最長(zhǎng)時(shí)間，防止 ddos  
  this.headersTimeout = 60 * 1000; 
}

function connectionListener(socket) {  
  defaultTriggerAsyncIdScope(  
    getOrSetAsyncId(socket), connectionListenerInternal, this, socket  
  );  
}  

// socket 表示新連接  
function connectionListenerInternal(server, socket) {  
  // socket 所屬 server  
  socket.server = server;  
  // 分配一個(gè) HTTP 解析器  
  const parser = parsers.alloc();  
  // 初始化解析器
  parser.initialize(HTTPParser.REQUEST, ...);  
  // 關(guān)聯(lián)起來
  parser.socket = socket;  
  socket.parser = parser;  

  const state = {  
    onData: null,  
    // 同一 TCP 連接上，請(qǐng)求和響應(yīng)的的隊(duì)列，線頭阻塞的原理  
    outgoing: [],  
    incoming: [],   
  };  
  // 監(jiān)聽 TCP 上的數(shù)據(jù)，開始解析 HTTP 報(bào)文  
  state.onData = socketOnData.bind(undefined, 
                                   server, 
                                   socket, 
                                   parser, 
                                   state);  
  socket.on('data', state.onData);
  // 解析 HTTP 頭部完成后執(zhí)行的回調(diào)  
  parser.onIncoming = parserOnIncoming.bind(undefined, 
                                            server, 
                                            socket, 
                                            state);  
  /*
    如果 handle 是繼承 StreamBase 的流，則在 C++ 層解析 HTTP 請(qǐng)求報(bào)文，
    否則使用上面的 socketOnData 函數(shù)處理 HTTP 請(qǐng)求報(bào)文，
    TCP 模塊的 isStreamBase 為 true 
  */
  if (socket._handle && socket._handle.isStreamBase &&  
      !socket._handle._consumed) {  
    parser._consumed = true;  
    socket._handle._consumed = true;  
    parser.consume(socket._handle);  
  }  
  // 執(zhí)行 llhttp_execute 時(shí)的回調(diào)
  parser[kOnExecute] = onParserExecute.bind(undefined, 
                                            server, 
                                            socket, 
                                            parser, 
                                            state);   
}

const parsers = new FreeList('parsers', 1000, function parsersCb() {
  const parser = new HTTPParser();

  cleanParser(parser);

  parser.onIncoming = null;
  // 各種鉤子毀掉
  parser[kOnHeaders] = parserOnHeaders;
  parser[kOnHeadersComplete] = parserOnHeadersComplete;
  parser[kOnBody] = parserOnBody;
  parser[kOnMessageComplete] = parserOnMessageComplete;

  return parser;
});

function socketOnData(server, socket, parser, state, d) {  
  // 交給 HTTP 解析器處理，返回已經(jīng)解析的字節(jié)數(shù)  
  const ret = parser.execute(d);  
}

if (socket._handle && socket._handle.isStreamBase) {  
    parser.consume(socket._handle);  
}

static void Consume(const FunctionCallbackInfo<Value>& args) {
    Parser* parser;
    ASSIGN_OR_RETURN_UNWRAP(&parser, args.Holder());
    // 解析出 C++ TCPWrap 對(duì)象
    StreamBase* stream = StreamBase::FromObject(args[0].As<Object>());
    // 注冊(cè) parser 成為流的消費(fèi)者，即 TCP 數(shù)據(jù)的消費(fèi)者
    stream->PushStreamListener(parser);
}

void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override {  
    // 解析 HTTP 協(xié)議
    Local<Value> ret = Execute(buf.base, nread);  
    // 執(zhí)行 kOnExecute 回調(diào)
    Local<Value> cb = object()->Get(env()->context(), kOnExecute).ToLocalChecked();  
    MakeCallback(cb.As<Function>(), 1, &ret);  
}

function parserOnHeaders(headers, url) {
  // 記錄解析到的 HTTP 頭
  if (this.maxHeaderPairs <= 0 ||
      this._headers.length < this.maxHeaderPairs) {
    this._headers = this._headers.concat(headers);
  }
  this._url += url;
}

function parserOnHeadersComplete(versionMajor, versionMinor, headers, method,
                                 url, statusCode, statusMessage, upgrade,
                                 shouldKeepAlive) {
  const parser = this;
  const { socket } = parser;
  // 創(chuàng)建一個(gè)對(duì)象表示收到的 HTTP 請(qǐng)求
  const ParserIncomingMessage = (socket && socket.server &&
                                 socket.server[kIncomingMessage]) ||
                                 IncomingMessage;
  // 新建一個(gè)IncomingMessage對(duì)象
  const incoming = parser.incoming = new ParserIncomingMessage(socket);
  // 執(zhí)行回調(diào)
  return parser.onIncoming(incoming, shouldKeepAlive);
}

function parserOnIncoming(server, socket, state, req, keepAlive) {  

  // 請(qǐng)求入隊(duì)（待處理的請(qǐng)求隊(duì)列）  
  state.incoming.push(req);  
  // 新建一個(gè)表示響應(yīng)的對(duì)象  
  const res = new server[kServerResponse](req);  
  /*
     socket 當(dāng)前已經(jīng)在處理其它請(qǐng)求的響應(yīng)，則先排隊(duì)，
     否則掛載響應(yīng)對(duì)象到 socket，作為當(dāng)前處理的響應(yīng)  
  */
  if (socket._httpMessage) {  
    state.outgoing.push(res);  
  } else {  
    res.assignSocket(socket);  
  }  

  // 響應(yīng)處理完畢后，需要做一些處理  
  res.on('finish', resOnFinish.bind(undefined, 
                                    req, 
                                    res, 
                                    socket, 
                                    state, 
                                    server));  
  // 觸發(fā) request 事件說明有請(qǐng)求到來  
  server.emit('request', req, res);  
}

function parserOnBody(b, start, len) {
  // IncomingMessage 對(duì)象，即 request 對(duì)象
  const stream = this.incoming;
  // Pretend this was the result of a stream._read call.
  if (len > 0 && !stream._dumped) {
    const slice = b.slice(start, start + len);
    const ret = stream.push(slice);
    if (!ret)
      readStop(this.socket);
  }
}

const server= http.createServer((request, response) => {  
  request.on('data', (chunk) => {  
   // 處理body  
  });  
  request.on('end', () => {  
   // body結(jié)束  
  });  
})

function listenIncluster(server, address, port, addressType,  
                         backlog, fd, exclusive, flags) {  

  const serverQuery = {  
    address: address,  
    port: port,  
    addressType: addressType,  
    fd: fd,  
    flags,  
  };  

  cluster._getServer(server, serverQuery, listenOnMasterHandle);    
  function listenOnMasterHandle(err, handle) {  
    server._handle = handle;  
    server._listen2(address,
                    port, 
                    addressType, 
                    backlog, 
                    fd, 
                    flags);  
  }  
}

cluster._getServer = function(obj, options, cb) {  
  let address = options.address;  
  const message = {  
    act: 'queryServer',  
    index,  
    data: null,  
    ...options  
  };  

  message.address = address;  
  // 給主進(jìn)程發(fā)送消息  
  send(message, (reply, handle) => {  
    // 根據(jù)不同模式做處理
    if (handle)  
      shared(reply, handle, indexesKey, cb);  
    else  
      rr(reply, indexesKey, cb);             
  });  
};

function queryServer(worker, message) {  
  const key = `${message.address}:${message.port}:${message.addressType}:${message.fd}:${message.index}`;  
  let handle = handles.get(key);  

  if (handle === undefined) {  
    let address = message.address;  
    let constructor = RoundRobinHandle;  
    // 根據(jù)策略選取不同的構(gòu)造函數(shù)，UDP 只能使用共享模式，因?yàn)?UDP 不是基于連接的，沒有連接可以分發(fā)  
    if (schedulingPolicy !== SCHED_RR ||  
        message.addressType === 'udp4' ||  
        message.addressType === 'udp6') {  
      constructor = SharedHandle;  
    }  

    handle = new constructor(key,  
                             address,  
                             message.port,  
                             message.addressType,  
                             message.fd,  
                             message.flags);  
    handles.set(key, handle);  
  }  
  handle.add(worker, (errno, reply, handle) => {  
    const { data } = handles.get(key);  
    // 返回結(jié)果給子進(jìn)程
    send(worker, {  
      errno,  
      key,  
      ack: message.seq,  
      data,  
      ...reply  
    }, handle);  
  });  
}

function SharedHandle(key, address, port, addressType, fd, flags) {  
  this.key = key;  
  this.workers = [];  
  this.handle = null;  
  this.errno = 0;  

  let rval;  
  if (addressType === 'udp4' || addressType === 'udp6')  
    rval = dgram._createSocketHandle(address, 
                                     port, 
                                     addressType, 
                                     fd, 
                                     flags);  
  else  
    rval = net._createServerHandle(address,  
                                   port, 
                                   addressType, 
                                   fd, 
                                   flags);  

  if (typeof rval === 'number')  
    this.errno = rval;  
  else  
    this.handle = rval;  
}

SharedHandle.prototype.add = function(worker, send) {  
  this.workers.push(worker);  
  send(this.errno, null, this.handle);  
};

function shared(message, handle, indexesKey, cb) {  
  const key = message.key;  

  const close = handle.close;  

  handle.close = function() {  
    send({ act: 'close', key });  
    handles.delete(key);  
    indexes.delete(indexesKey);  
    // 因?yàn)槭枪蚕淼?，可以直?close 掉而不會(huì)影響其它子進(jìn)程等
    return close.apply(handle, arguments);  
  };  
  handles.set(key, handle); 
  // 執(zhí)行 net 模塊的回調(diào) 
  cb(message.errno, handle);  
}

function setupListenHandle(address, port, addressType, backlog, fd, flags) {
  // this._handle 即主進(jìn)程返回的 handle
  // 連接到來時(shí)的回調(diào)
  this._handle.onconnection = onconnection;
  this._handle[owner_symbol] = this;
  const err = this._handle.listen(backlog || 511);
}

const cluster = require('cluster');
const os = require('os');
// 設(shè)置為共享模式
cluster.schedulingPolicy = cluster.SCHED_NONE;

// 主進(jìn)程 fork 多個(gè)子進(jìn)程
if (cluster.isMaster) {
  // 通常根據(jù) CPU 核數(shù)創(chuàng)建多個(gè)進(jìn)程 os.cpus().length
  for (let i = 0; i < 3; i++) {
    cluster.fork();
  }
} else { // 子進(jìn)程創(chuàng)建服務(wù)器
  const net = require('net');
  const server = net.createServer((socket) => {
    socket.destroy();
    console.log(`handled by process: ${process.pid}`);
  });
  server.listen(8080);
}

function RoundRobinHandle(key, address, port, addressType, fd, flags) {  
  this.key = key;  
  this.all = new Map();  
  this.free = [];  
  this.handles = [];  
  this.handle = null;  
  this.server = net.createServer(assert.fail);  

  if (fd >= 0)  
    this.server.listen({ fd });  
  else if (port >= 0) {  
    // 啟動(dòng)一個(gè)服務(wù)器
    this.server.listen({  
      port,  
      host: address,  
      ipv6Only: Boolean(flags & constants.UV_TCP_IPV6ONLY),  
    });  
  } else  
    this.server.listen(address);  // UNIX socket path.  
  // 監(jiān)聽成功后，注冊(cè) onconnection 回調(diào)，有連接到來時(shí)執(zhí)行  
  this.server.once('listening', () => {  
    this.handle = this.server._handle;  
    // 分發(fā)請(qǐng)求給子進(jìn)程
    this.handle.onconnection = (err, handle) => this.distribute(err, handle);  
    this.server._handle = null;  
    this.server = null;  
  });  
}

RoundRobinHandle.prototype.add = function(worker, send) {  
   this.all.set(worker.id, worker);  

   const done = () => {  
    // send 的第三個(gè)參數(shù)是 null，說明沒有 handle
    if (this.handle.getsockname) {  
      const out = {};  
      this.handle.getsockname(out);  
      send(null, { sockname: out }, null);  
    } else {  
      send(null, null, null);  // UNIX socket.  
    }  

    this.handoff(worker);   
  };  
  // 否則等待 listen 成功后執(zhí)行回調(diào)  
  this.server.once('listening', done);  
};

handle.add(worker, (errno, reply, handle) => {  
  const { data } = handles.get(key);  
  send(worker, {  
    errno,  
    key,  
    ack: message.seq,  
    data,  
    ...reply  
  }, handle);  
});

function rr(message, indexesKey, cb) { 
   let key = message.key;  
   // 不需要 listen，空操作
   function listen(backlog) {  
       return 0;  
   }  

   function close() {
       // 因?yàn)?handle 是共享的，所以無法直接關(guān)閉，需要告訴父進(jìn)程，引用數(shù)減一
       if (key === undefined)
         return;

       send({ act: 'close', key });
       handles.delete(key);
       indexes.delete(indexesKey);
       key = undefined;
  } 
  // 構(gòu)造假的 handle 給調(diào)用方
  const handle = { close, listen, ref: noop, unref: noop };  

  handles.set(key, handle); 
  // 執(zhí)行 net 模塊的回調(diào) 
  cb(0, handle);  
}

RoundRobinHandle.prototype.distribute = function(err, handle) {  
  // 首先保存 handle 到隊(duì)列  
  this.handles.push(handle);  
  // 從空閑隊(duì)列獲取一個(gè)子進(jìn)程  
  const worker = this.free.shift();  
  // 分發(fā)  
  if (worker)  
    this.handoff(worker);  
};  

RoundRobinHandle.prototype.handoff = function(worker) {  
  // 拿到一個(gè) handle  
  const handle = this.handles.shift();  
  // 沒有 handle，則子進(jìn)程重新入隊(duì)  
  if (handle === undefined) {  
    this.free.push(worker);
    return;  
  }  
  // 通知子進(jìn)程有新連接  
  const message = { act: 'newconn', key: this.key };  

  sendHelper(worker.process, message, handle, (reply) => {  
    // 接收成功  
    if (reply.accepted)  
      handle.close();  
    else  
      // 結(jié)束失敗，則重新分發(fā)  
      this.distribute(0, handle);
    // 繼續(xù)分發(fā)
    this.handoff(worker);  
  });  
};

function onmessage(message, handle) {  
    if (message.act === 'newconn')  
      onconnection(message, handle);  
}  

function onconnection(message, handle) {  
  const key = message.key;  
  const server = handles.get(key);  
  const accepted = server !== undefined;  
  // 回復(fù)接收成功  
  send({ ack: message.seq, accepted });  

  if (accepted)  
     // 在 net 模塊設(shè)置
    server.onconnection(0, handle);  
}

const cluster = require('cluster');
const os = require('os');
// 設(shè)置為輪詢模式
cluster.schedulingPolicy = cluster.SCHED_RR;

// 主進(jìn)程 fork 多個(gè)子進(jìn)程
if (cluster.isMaster) {
  // 通常根據(jù) CPU 核數(shù)創(chuàng)建多個(gè)進(jìn)程 os.cpus().length
  for (let i = 0; i < 3; i++) {
    cluster.fork();
  }
} else { // 子進(jìn)程創(chuàng)建服務(wù)器
  const net = require('net');
  const server = net.createServer((socket) => {
    socket.destroy();
    console.log(`handled by process: ${process.pid}`);
  });
  server.listen(8080);
}