[原]Redis源代码分析之四：Unix底层网络通信

winston

还是先介绍接口：int anetTcpConnect(char *err, char *addr, int port);

2. int anetTcpNonBlockConnect(char *err, char *addr, int port);
3. int anetUnixConnect(char *err, char *path);
4. int anetUnixNonBlockConnect(char *err, char *path);
5. int anetRead(int fd, char *buf, int count);
6. int anetResolve(char *err, char *host, char *ipbuf);
7. int anetTcpServer(char *err, int port, char *bindaddr);
8. int anetUnixServer(char *err, char *path, mode_t perm);
9. int anetTcpAccept(char *err, int serversock, char *ip, int *port);
10. int anetUnixAccept(char *err, int serversock);
11. int anetWrite(int fd, char *buf, int count);
12. int anetNonBlock(char *err, int fd);
13. int anetTcpNoDelay(char *err, int fd);
14. int anetTcpKeepAlive(char *err, int fd);
15. int anetPeerToString(int fd, char *ip, int *port);

复制代码

从Anet接口定义看出，API主要为对Unix网络通信API的抽象和封装。从网络类型，主要分为Tcp和Unix两种；从同步模型，主要分为Block和NonBlock两种。通信操作包括Connect、Accept、Read、Write、Resolve等，还有完整的Server。
下面我们以TCP为例重点分析。先看TcpConnect的实现函数anetTcpGenericConnect：static int anetTcpGenericConnect(char *err, char *addr, int port, int flags)

2. {
3.     int s;
4.     struct sockaddr_in sa;
5.     if ((s = anetCreateSocket(err,AF_INET)) == ANET_ERR)
6.         return ANET_ERR;
8.     sa.sin_family = AF_INET;
9.     sa.sin_port = htons(port);
10.     if (inet_aton(addr, &sa.sin_addr) == 0) {
11.         struct hostent *he;
13.         he = gethostbyname(addr);
14.         if (he == NULL) {
15.             anetSetError(err, "can't resolve: %s", addr);
16.             close(s);
17.             return ANET_ERR;
18.         }
19.         memcpy(&sa.sin_addr, he->h_addr, sizeof(struct in_addr));
20.     }
21.     if (flags & ANET_CONNECT_NONBLOCK) {
22.         if (anetNonBlock(err,s) != ANET_OK)
23.             return ANET_ERR;
24.     }
25.     if (connect(s, (struct sockaddr*)&sa, sizeof(sa)) == -1) {
26.         if (errno == EINPROGRESS &&
27.             flags & ANET_CONNECT_NONBLOCK)
28.             return s;
30.         anetSetError(err, "connect: %s", strerror(errno));
31.         close(s);
32.         return ANET_ERR;
33.     }
34.     return s;
35. }

复制代码

首先调用anetCreateSocket函数创建套接字：static int anetCreateSocket(char *err, int domain) {

2.     int s, on = 1;
3.     if ((s = socket(domain, SOCK_STREAM, 0)) == -1) {
4.         anetSetError(err, "creating socket: %s", strerror(errno));
5.         return ANET_ERR;
6.     }
8.     /* Make sure connection-intensive things like the redis benckmark
9.      * will be able to close/open sockets a zillion of times */
10.     if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) {
11.         anetSetError(err, "setsockopt SO_REUSEADDR: %s", strerror(errno));
12.         return ANET_ERR;
13.     }
14.     return s;
15. }

复制代码

SOCK_STREAM提供面向连接的稳定数据传输。后面设置套接字选项：


int setsockopt(int sockfd, int level, int optname, const void *optval, socklen_t optlen);因为是在套接字API的level设置套接字选项，所以level参数设为SOL_SOCKET。将SO_REUSEADDR设为开关on的值，以便在连接密集型的redis程序中频繁打开/关闭套接字时重用本地地址和端口。

下面填充sockaddress_in结构体，并发起连接。如果选择非阻塞方式，那么将调用anetNonBlock函数：int anetNonBlock(char *err, int fd)

2. {
3.     int flags;
5.     /* Set the socket nonblocking.
6.      * Note that fcntl(2) for F_GETFL and F_SETFL can't be
7.      * interrupted by a signal. */
8.     if ((flags = fcntl(fd, F_GETFL)) == -1) {
9.         anetSetError(err, "fcntl(F_GETFL): %s", strerror(errno));
10.         return ANET_ERR;
11.     }
12.     if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
13.         anetSetError(err, "fcntl(F_SETFL,O_NONBLOCK): %s", strerror(errno));
14.         return ANET_ERR;
15.     }
16.     return ANET_OK;
17. }

复制代码

我们看到，首先通过fcntl(fd, F_GETFL)获得描述符状态标识，然后通过fcntl(fd, F_SETFL, flags | O_NONBLOCK)将非阻塞属性增加到前面读来的flags中。

注意如果采用了非阻塞方式，返回错误码EINPROGRESS并不算错误，只是表明连接正在进行中；而在阻塞方式连接中，就应当按做错误处理。
anetUnixGenericConnect函数与anetTCPGenericConnect函数类似，只是地址描述结构不是sockaddr_in而是sockaddr_un。



下面分析另一个典型函数anetTcpServer：int anetTcpServer(char *err, int port, char *bindaddr)

2. {
3.     int s;
4.     struct sockaddr_in sa;
6.     if ((s = anetCreateSocket(err,AF_INET)) == ANET_ERR)
7.         return ANET_ERR;
9.     memset(&sa,0,sizeof(sa));
10.     sa.sin_family = AF_INET;
11.     sa.sin_port = htons(port);
12.     sa.sin_addr.s_addr = htonl(INADDR_ANY);
13.     if (bindaddr && inet_aton(bindaddr, &sa.sin_addr) == 0) {
14.         anetSetError(err, "invalid bind address");
15.         close(s);
16.         return ANET_ERR;
17.     }
18.     if (anetListen(err,s,(struct sockaddr*)&sa,sizeof(sa)) == ANET_ERR)
19.         return ANET_ERR;
20.     return s;
21. }

复制代码

TCP服务器的处理流程基本是：创建套接字，初始化地址结构，然后调用anetListene进入监听。监听也分两个步骤，首先把套接字和地址结构绑定的bind，然后开启监听。
服务器创建成功后，返回套接字描述符。然后调用anetGenericAccept开始进入accept状态：static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {

2.     int fd;
3.     while(1) {
4.         fd = accept(s,sa,len);
5.         if (fd == -1) {
6.             if (errno == EINTR)
7.                 continue;
8.             else {
9.                 anetSetError(err, "accept: %s", strerror(errno));
10.                 return ANET_ERR;
11.             }
12.         }
13.         break;
14.     }
15.     return fd;
16. }

复制代码

客户端发起的连接一旦被accept，就可以读写数据了：/* Like read(2) but make sure 'count' is read before to return

2. * (unless error or EOF condition is encountered) */
3. int anetRead(int fd, char *buf, int count)
4. {
5.     int nread, totlen = 0;
6.     while(totlen != count) {
7.         nread = read(fd,buf,count-totlen);
8.         if (nread == 0) return totlen;
9.         if (nread == -1) return -1;
10.         totlen += nread;
11.         buf += nread;
12.     }
13.     return totlen;
14. }
16. /* Like write(2) but make sure 'count' is read before to return
17. * (unless error is encountered) */
18. int anetWrite(int fd, char *buf, int count)
19. {
20.     int nwritten, totlen = 0;
21.     while(totlen != count) {
22.         nwritten = write(fd,buf,count-totlen);
23.         if (nwritten == 0) return totlen;
24.         if (nwritten == -1) return -1;
25.         totlen += nwritten;
26.         buf += nwritten;
27.     }
28.     return totlen;
29. }

复制代码

作者：Aegeaner 发表于2012-2-4 14:29:33 原文链接