站長資訊網本篇文章帶大家聊聊Node.js中的網絡與流，涉及的知識點有ibuv中網絡的實現、BSD 套接字、UNIX 域協議使用等，下面一起來看看吧！
【推薦學習：《nodejs 教程》】
本篇例子來源：http://docs.libuv.org/en/v1.x/guide/networking.html
涉及的知識點
- libuv 中網絡的實現
- libuv 解決 accept （EMFILE錯誤）
- BSD 套接字
- SOCKADDR_IN
- UNIX 域協議使用！ 在進程間傳遞“文件描述符”
例子 tcp-echo-server/main.c
libuv 異步使用 BSD 套接字 的例子
libuv 中的網絡和直接使用 BSD 套接字接口沒有什么不同，有些事情更簡單，都是無阻塞的，但概念都是一樣的。此外，libuv 還提供了一些實用的函數來抽象出那些煩人的、重復的、低級的任務，比如使用BSD套接字結構設置套接字、DNS查詢以及調整各種套接字參數。
int main() { loop = uv_default_loop(); uv_tcp_t server; uv_tcp_init(loop, &server); uv_ip4_addr("0.0.0.0", DEFAULT_PORT, &addr); uv_tcp_bind(&server, (const struct sockaddr*)&addr, 0); int r = uv_listen((uv_stream_t*) &server, DEFAULT_BACKLOG, on_new_connection); if (r) { fprintf(stderr, "Listen error %sn", uv_strerror(r)); return 1; } return uv_run(loop, UV_RUN_DEFAULT); } void on_new_connection(uv_stream_t *server, int status) { if (status < 0) { fprintf(stderr, "New connection error %sn", uv_strerror(status)); // error! return; } uv_tcp_t *client = (uv_tcp_t*) malloc(sizeof(uv_tcp_t)); uv_tcp_init(loop, client); if (uv_accept(server, (uv_stream_t*) client) == 0) { uv_read_start((uv_stream_t*) client, alloc_buffer, echo_read); }
同步的例子
這是一個正常同步使用 BSD 套接字 的例子。
作為參照可以發現主要有如下幾步
-
首先調用 socket() 為通訊創建一個端點，為套接字返回一個文件描述符。
-
接著調用 bind() 為一個套接字分配地址。當使用 socket() 創建套接字后，只賦予其所使用的協議，并未分配地址。在接受其它主機的連接前，必須先調用 bind() 為套接字分配一個地址。
-
當 socket 和一個地址綁定之后，再調用 listen() 函數會開始監聽可能的連接請求。
-
最后調用 accept, 當應用程序監聽來自其他主機的面對數據流的連接時，通過事件（比如Unix select()系統調用）通知它。必須用 accept()函數初始化連接。 accept() 為每個連接創立新的套接字并從監聽隊列中移除這個連接。
int main(void) { struct sockaddr_in stSockAddr; int SocketFD = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP); if(-1 == SocketFD) { perror("can not create socket"); exit(EXIT_FAILURE); } memset(&stSockAddr, 0, sizeof(struct sockaddr_in)); stSockAddr.sin_family = AF_INET; stSockAddr.sin_port = htons(1100); stSockAddr.sin_addr.s_addr = INADDR_ANY; if(-1 == bind(SocketFD,(const struct sockaddr *)&stSockAddr, sizeof(struct sockaddr_in))) { perror("error bind failed"); close(SocketFD); exit(EXIT_FAILURE); } if(-1 == listen(SocketFD, 10)) { perror("error listen failed"); close(SocketFD); exit(EXIT_FAILURE); } for(;;) { int ConnectFD = accept(SocketFD, NULL, NULL); if(0 > ConnectFD) { perror("error accept failed"); close(SocketFD); exit(EXIT_FAILURE); } /* perform read write operations ... */ shutdown(ConnectFD, SHUT_RDWR); close(ConnectFD); } close(SocketFD); return 0; }
uv_tcp_init
main > uv_tcp_init
1、對 domain 進行了驗證, 需要是下面3種的一種
- AF_INET 表示 IPv4 網絡協議
- AF_INET6 表示 IPv6
- AF_UNSPEC 表示適用于指定主機名和服務名且適合任何協議族的地址
2、tcp 也是一種流, 調用 uv__stream_init 對流數據進行初始化
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 = flags & 0xFF; if (domain != AF_INET && domain != AF_INET6 && domain != AF_UNSPEC) return UV_EINVAL; if (flags & ~0xFF) return UV_EINVAL; uv__stream_init(loop, (uv_stream_t*)tcp, UV_TCP); ... return 0; }
uv__stream_init
main > uv_tcp_init > uv__stream_init
流的初始化函數使用的地方還是特別多的, 也特別重要。下述 i/o 的完整實現參考 【libuv 源碼學習筆記】線程池與i/o
1、對流會被調用的回調函數等進行一個初始化
- 如 read_cb 函數, 在本例子中 on_new_connection > uv_read_start 函數就會真實的設置該 read_cb 為用戶傳入的參數 echo_read, 其被調用時機是該 stream 上設置的 io_watcher.fd 有數據寫入時, 在事件循環階段被 epoll 捕獲后。
- alloc_cb 函數的調用過程同 read_cb, alloc 類型函數一般是設置當前需要讀取的內容長度, 在流數據傳輸時通常首先會寫入本次傳輸數據的長度, 然后是具體的內容, 主要是為了接收方能夠合理的申請內存進行存儲。如 grpc, thread-loader 中都有詳細的應用。
- close_cb 函數被調用在 stream 數據結束時或者出錯時。
- connection_cb 函數如本例子 tcp 流, 當 accept 接收到新連接時被調用。本例子中即為 on_new_connection
- connect_req 結構主要用于 tcp 客戶端相關連接回調等數據的掛載使用。
- shutdown_req 結構主要用于流 destroy 時回調等數據的掛載使用。
- accepted_fd 當 accept 接收到新連接時, 存儲 accept(SocketFD, NULL, NULL) 返回的 ConnectFD。
- queued_fds 用于保存等待處理的連接, 其主要用于 node cluster 集群 的實現。
// queued_fds 1. 當收到其他進程通過 ipc 寫入的數據時, 調用 uv__stream_recv_cmsg 函數 2. uv__stream_recv_cmsg 函數讀取到進程傳遞過來的 fd 引用, 調用 uv__stream_queue_fd 函數保存。 3. queued_fds 被消費主要在 src/stream_wrap.cc LibuvStreamWrap::OnUvRead > AcceptHandle 函數中。
2、其中專門為 loop->emfile_fd 通過 uv__open_cloexec 方法創建一個指向空文件(/dev/null)的 idlefd 文件描述符, 追蹤發現原來是解決 accept （EMFILE錯誤）, 下面我們講 uv__accept 的時候再細說這個 loop->emfile_fd 的妙用。
accept處理連接時，若出現 EMFILE 錯誤不進行處理，則內核間隔性嘗試連接，導致整個網絡設計程序崩潰
3、調用 uv__io_init 初始化的該 stream 的 i/o 觀察者的回調函數為 uv__stream_io
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_) */ uv__io_init(&stream->io_watcher, uv__stream_io, -1); }
uv__open_cloexec
main > uv_tcp_init > uv__stream_init > uv__open_cloexec
同步調用 open 方法拿到了 fd, 也許你會問為啥不像 【libuv 源碼學習筆記】線程池與i/o 中調用 uv_fs_open 異步獲取 fd， 其實 libuv 中并不全部是異步的實現, 比如當前的例子啟動 tcp 服務前的一些初始化, 而不是用戶請求過程中發生的任務, 同步也是能接受的。
int uv__open_cloexec(const char* path, int flags) { #if defined(O_CLOEXEC) int fd; fd = open(path, flags | O_CLOEXEC); if (fd == -1) return UV__ERR(errno); return fd; #else /* O_CLOEXEC */ int err; int fd; fd = open(path, flags); if (fd == -1) return UV__ERR(errno); err = uv__cloexec(fd, 1); if (err) { uv__close(fd); return err; } return fd; #endif /* O_CLOEXEC */ }
uv__stream_io
main > uv_tcp_init > uv__stream_init > uv__stream_io
雙工流的 i/o 觀察者回調函數, 如調用的 stream->connect_req 函數, 其值是例子中 uv_listen 函數的最后一個參數 on_new_connection。
-
當發生 POLLIN | POLLERR | POLLHUP 事件時: 該 fd 有可讀數據時調用 uv__read 函數
-
當發生 POLLOUT | POLLERR | POLLHUP 事件時: 該 fd 有可讀數據時調用 uv__write 函數
static void uv__stream_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) { uv_stream_t* stream; stream = container_of(w, uv_stream_t, io_watcher); assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE || stream->type == UV_TTY); assert(!(stream->flags & UV_HANDLE_CLOSING)); if (stream->connect_req) { uv__stream_connect(stream); return; } assert(uv__stream_fd(stream) >= 0); if (events & (POLLIN | POLLERR | POLLHUP)) uv__read(stream); if (uv__stream_fd(stream) == -1) return; /* read_cb closed stream. */ if ((events & POLLHUP) && (stream->flags & UV_HANDLE_READING) && (stream->flags & UV_HANDLE_READ_PARTIAL) && !(stream->flags & UV_HANDLE_READ_EOF)) { uv_buf_t buf = { NULL, 0 }; uv__stream_eof(stream, &buf); } if (uv__stream_fd(stream) == -1) return; /* read_cb closed stream. */ if (events & (POLLOUT | POLLERR | POLLHUP)) { uv__write(stream); uv__write_callbacks(stream); /* Write queue drained. */ if (QUEUE_EMPTY(&stream->write_queue)) uv__drain(stream); } }
uv_ip4_addr
main > uv_ip4_addr
uv_ip4_addr 用于將人類可讀的 IP 地址、端口對轉換為 BSD 套接字 API 所需的 sockaddr_in 結構。
int uv_ip4_addr(const char* ip, int port, struct sockaddr_in* addr) { memset(addr, 0, sizeof(*addr)); addr->sin_family = AF_INET; addr->sin_port = htons(port); #ifdef SIN6_LEN addr->sin_len = sizeof(*addr); #endif return uv_inet_pton(AF_INET, ip, &(addr->sin_addr.s_addr)); }
uv_tcp_bind
main > uv_tcp_bind
從 uv_ip4_addr 函數的實現, 其實是在 addr 的 sin_family 上面設置值為 AF_INET, 但在 uv_tcp_bind 函數里面卻是從 addr 的 sa_family屬性上面取的值, 這讓 c 初學者的我又陷入了一陣思考 …
sockaddr_in 和 sockaddr 是并列的結構，指向 sockaddr_in 的結構體的指針也可以指向 sockaddr 的結構體，并代替它。也就是說，你可以使用 sockaddr_in 建立你所需要的信息,然后用 memset 函數初始化就可以了memset((char*)&mysock,0,sizeof(mysock));//初始化
原來是這樣, 這里通過強制指針類型轉換 const struct sockaddr* addr 達到的目的, 函數的最后調用了 uv__tcp_bind 函數。
int uv_tcp_bind(uv_tcp_t* handle, const struct sockaddr* addr, unsigned int flags) { unsigned int addrlen; if (handle->type != UV_TCP) return UV_EINVAL; if (addr->sa_family == AF_INET) addrlen = sizeof(struct sockaddr_in); else if (addr->sa_family == AF_INET6) addrlen = sizeof(struct sockaddr_in6); else return UV_EINVAL; return uv__tcp_bind(handle, addr, addrlen, flags); }
uv__tcp_bind
main > uv_tcp_bind > uv__tcp_bind
-
調用 maybe_new_socket, 如果當前未設置 socketfd, 則調用 new_socket 獲取
-
調用 setsockopt 用于為指定的套接字設定一個特定的套接字選項
-
調用 bind 為一個套接字分配地址。當使用socket()創建套接字后，只賦予其所使用的協議，并未分配地址。
int uv__tcp_bind(uv_tcp_t* tcp, const struct sockaddr* addr, unsigned int addrlen, unsigned int flags) { int err; int on; /* Cannot set IPv6-only mode on non-IPv6 socket. */ if ((flags & UV_TCP_IPV6ONLY) && addr->sa_family != AF_INET6) return UV_EINVAL; 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); ... errno = 0; if (bind(tcp->io_watcher.fd, addr, addrlen) && errno != EADDRINUSE) { if (errno == EAFNOSUPPORT) return UV_EINVAL; return UV__ERR(errno); } ... }
new_socket
main > uv_tcp_bind > uv__tcp_bind > maybe_new_socket > new_socket
-
通過 uv__socket 其本質調用 socket 獲取到 sockfd
-
調用 uv__stream_open 設置 stream i/o 觀察的 fd 為步驟1 拿到的 sockfd
static int new_socket(uv_tcp_t* handle, int domain, unsigned long flags) { struct sockaddr_storage saddr; socklen_t slen; int sockfd; int err; err = uv__socket(domain, SOCK_STREAM, 0); if (err < 0) return err; sockfd = err; err = uv__stream_open((uv_stream_t*) handle, sockfd, flags); ... return 0; }
uv__stream_open
main > uv_tcp_bind > uv__tcp_bind > maybe_new_socket > new_socket > uv__stream_open
主要用于設置 stream->io_watcher.fd 為參數傳入的 fd。
int uv__stream_open(uv_stream_t* stream, int fd, int flags) { #if defined(__APPLE__) int enable; #endif 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); } } #if defined(__APPLE__) enable = 1; if (setsockopt(fd, SOL_SOCKET, SO_OOBINLINE, &enable, sizeof(enable)) && errno != ENOTSOCK && errno != EINVAL) { return UV__ERR(errno); } #endif stream->io_watcher.fd = fd; return 0; }
uv_listen
main > uv_listen
主要調用了 uv_tcp_listen 函數。
int uv_listen(uv_stream_t* stream, int backlog, uv_connection_cb cb) { int err; err = ERROR_INVALID_PARAMETER; switch (stream->type) { case UV_TCP: err = uv_tcp_listen((uv_tcp_t*)stream, backlog, cb); break; case UV_NAMED_PIPE: err = uv_pipe_listen((uv_pipe_t*)stream, backlog, cb); break; default: assert(0); } return uv_translate_sys_error(err); }
uv_tcp_listen
main > uv_listen > uv_tcp_listen
-
調用 listen 開始監聽可能的連接請求
-
掛載例子中傳入的回調 on_new_connection
-
暴力改寫 i/o 觀察者的回調, 在上面的 uv__stream_init 函數中, 通過 uv__io_init 設置了 i/o 觀察者的回調為 uv__stream_io, 作為普通的雙工流是適用的, 這里 tcp 流直接通過 tcp->io_watcher.cb = uv__server_io 賦值語句設置 i/o 觀察者回調為 uv__server_io
-
調用 uv__io_start 注冊 i/o 觀察者, 開始監聽工作。
int uv_tcp_listen(uv_tcp_t* tcp, int backlog, uv_connection_cb cb) { ... if (listen(tcp->io_watcher.fd, backlog)) return UV__ERR(errno); tcp->connection_cb = cb; tcp->flags |= UV_HANDLE_BOUND; /* Start listening for connections. */ tcp->io_watcher.cb = uv__server_io; uv__io_start(tcp->loop, &tcp->io_watcher, POLLIN); return 0; }
uv__server_io
main > uv_listen > uv_tcp_listen > uv__server_io
tcp 流的 i/o 觀察者回調函數
-
調用 uv__accept, 拿到該連接的 ConnectFD
-
此時如果出現了上面 uv__stream_init 時說的 accept （EMFILE錯誤）, 則調用 uv__emfile_trick 函數
-
把步驟1拿到的 ConnectFD 掛載在了 stream->accepted_fd 上面
-
調用例子中傳入的回調 on_new_connection
void uv__server_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) { ... while (uv__stream_fd(stream) != -1) { assert(stream->accepted_fd == -1); err = uv__accept(uv__stream_fd(stream)); if (err < 0) { if (err == UV_EAGAIN || err == UV__ERR(EWOULDBLOCK)) return; /* Not an error. */ if (err == UV_ECONNABORTED) continue; /* Ignore. Nothing we can do about that. */ if (err == UV_EMFILE || err == UV_ENFILE) { err = uv__emfile_trick(loop, uv__stream_fd(stream)); if (err == UV_EAGAIN || err == UV__ERR(EWOULDBLOCK)) break; } stream->connection_cb(stream, err); continue; } UV_DEC_BACKLOG(w) stream->accepted_fd = err; stream->connection_cb(stream, 0); ... }
uv__emfile_trick
main > uv_listen > uv_tcp_listen > uv__server_io > uv__emfile_trick
在上面的 uv__stream_init 函數中, 我們發現 loop 的 emfile_fd 屬性上通過 uv__open_cloexec 方法創建一個指向空文件(/dev/null)的 idlefd 文件描述符。
當出現 accept （EMFILE錯誤）即文件描述符用盡時的錯誤時
首先將 loop->emfile_fd 文件描述符, 使其能 accept 新連接, 然后我們新連接將其關閉，以使其低于EMFILE的限制。接下來，我們接受所有等待的連接并關閉它們以向客戶發出信號，告訴他們我們已經超載了–我們確實超載了，但是我們仍在繼續工作。
static int uv__emfile_trick(uv_loop_t* loop, int accept_fd) { int err; int emfile_fd; if (loop->emfile_fd == -1) return UV_EMFILE; uv__close(loop->emfile_fd); loop->emfile_fd = -1; do { err = uv__accept(accept_fd); if (err >= 0) uv__close(err); } while (err >= 0 || err == UV_EINTR); emfile_fd = uv__open_cloexec("/", O_RDONLY); if (emfile_fd >= 0) loop->emfile_fd = emfile_fd; return err; }
on_new_connection
當收到一個新連接, 例子中的 on_new_connection 函數被調用
-
通過 uv_tcp_init 初始化了一個 tcp 客戶端流
-
調用 uv_accept 函數
void on_new_connection(uv_stream_t *server, int status) { if (status < 0) { fprintf(stderr, "New connection error %sn", uv_strerror(status)); // error! return; } uv_tcp_t *client = (uv_tcp_t*) malloc(sizeof(uv_tcp_t)); uv_tcp_init(loop, client); if (uv_accept(server, (uv_stream_t*) client) == 0) { uv_read_start((uv_stream_t*) client, alloc_buffer, echo_read); }
uv_accept
on_new_connection > uv_accept
根據不同的協議調用不同的方法, 該例子 tcp 調用 uv__stream_open 方法
uv__stream_open 設置給初始化完成的 client 流設置了 i/o 觀察者的 fd。該 fd 即是 uv__server_io 中提到的 ConnectFD 。
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: err = uv__stream_open(client, server->accepted_fd, UV_HANDLE_READABLE | UV_HANDLE_WRITABLE); if (err) { /* TODO handle error */ uv__close(server->accepted_fd); goto done; } break; case UV_UDP: 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: /* Process queued fds */ if (server->queued_fds != NULL) { uv__stream_queued_fds_t* queued_fds; queued_fds = server->queued_fds; /* Read first */ 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
on_new_connection > uv_read_start
開啟一個流的監聽工作
-
掛載回調函數 read_cb 為例子中的 echo_read, 當流有數據寫入時被調用
-
掛載回調函數 alloc_cb 為例子中的 alloc_buffer
-
調用 uv__io_start 函數, 這可是老朋友了, 通常用在 uv__io_init 初始化 i/o 觀察者后面, 用于注冊 i/o 觀察者。
uv_read_start 主要是調用了 uv__read_start 函數。開始了普通流的 i/o 過程。
- 初始化 i/o 觀察者在 uv_tcp_init > uv_tcp_init_ex > uv__stream_init > uv__io_init 設置其觀察者回調函數為 uv__stream_io
- 注冊 i/o 觀察者為 uv__io_start 開始監聽工作。
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); /* The UV_HANDLE_READING flag is irrelevant of the state of the tcp - it just * expresses the desired state of the user. */ stream->flags |= UV_HANDLE_READING; /* TODO: try to do the read inline? */ /* TODO: keep track of tcp state. If we've gotten a EOF then we should * not start the IO watcher. */ assert(uv__stream_fd(stream) >= 0); assert(alloc_cb); stream->read_cb = read_cb; stream->alloc_cb = alloc_cb; uv__io_start(stream->loop, &stream->io_watcher, POLLIN); uv__handle_start(stream); uv__stream_osx_interrupt_select(stream); return 0; }
小結
- uv_tcp_init 初始化 TCP Server handle, 其綁定的 fd 為 socket 返回的 socketFd。
- uv_tcp_bind 調用 bind 為套接字分配一個地址
- uv_listen 調用 listen 開始監聽可能的連接請求
- uv_accept 調用 accept 去接收一個新連接
- uv_tcp_init 初始化 TCP Client handle, 其綁定的 fd 為 accept 返回的 acceptFd, 剩下的就是一個普通流的讀寫 i/o 觀察。
原文地址：https://juejin.cn/post/6982226661081088036
作者：多小凱
