找回密码
 用户注册

QQ登录

只需一步,快速开始

查看: 5311|回复: 0

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

[复制链接]
发表于 2012-2-4 21:55:30 | 显示全部楼层 |阅读模式
还是先介绍接口int anetTcpConnect(char *err, char *addr, int port);
  1. int anetTcpNonBlockConnect(char *err, char *addr, int port);
  2. int anetUnixConnect(char *err, char *path);
  3. int anetUnixNonBlockConnect(char *err, char *path);
  4. int anetRead(int fd, char *buf, int count);
  5. int anetResolve(char *err, char *host, char *ipbuf);
  6. int anetTcpServer(char *err, int port, char *bindaddr);
  7. int anetUnixServer(char *err, char *path, mode_t perm);
  8. int anetTcpAccept(char *err, int serversock, char *ip, int *port);
  9. int anetUnixAccept(char *err, int serversock);
  10. int anetWrite(int fd, char *buf, int count);
  11. int anetNonBlock(char *err, int fd);
  12. int anetTcpNoDelay(char *err, int fd);
  13. int anetTcpKeepAlive(char *err, int fd);
  14. 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)
  1. {
  2.     int s;
  3.     struct sockaddr_in sa;
  4.     if ((s = anetCreateSocket(err,AF_INET)) == ANET_ERR)
  5.         return ANET_ERR;
  6.     sa.sin_family = AF_INET;
  7.     sa.sin_port = htons(port);
  8.     if (inet_aton(addr, &sa.sin_addr) == 0) {
  9.         struct hostent *he;
  10.         he = gethostbyname(addr);
  11.         if (he == NULL) {
  12.             anetSetError(err, "can't resolve: %s", addr);
  13.             close(s);
  14.             return ANET_ERR;
  15.         }
  16.         memcpy(&sa.sin_addr, he->h_addr, sizeof(struct in_addr));
  17.     }
  18.     if (flags & ANET_CONNECT_NONBLOCK) {
  19.         if (anetNonBlock(err,s) != ANET_OK)
  20.             return ANET_ERR;
  21.     }
  22.     if (connect(s, (struct sockaddr*)&sa, sizeof(sa)) == -1) {
  23.         if (errno == EINPROGRESS &&
  24.             flags & ANET_CONNECT_NONBLOCK)
  25.             return s;
  26.         anetSetError(err, "connect: %s", strerror(errno));
  27.         close(s);
  28.         return ANET_ERR;
  29.     }
  30.     return s;
  31. }
复制代码
首先调用anetCreateSocket函数创建套接字:static int anetCreateSocket(char *err, int domain) {
  1.     int s, on = 1;
  2.     if ((s = socket(domain, SOCK_STREAM, 0)) == -1) {
  3.         anetSetError(err, "creating socket: %s", strerror(errno));
  4.         return ANET_ERR;
  5.     }
  6.     /* Make sure connection-intensive things like the redis benckmark
  7.      * will be able to close/open sockets a zillion of times */
  8.     if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) {
  9.         anetSetError(err, "setsockopt SO_REUSEADDR: %s", strerror(errno));
  10.         return ANET_ERR;
  11.     }
  12.     return s;
  13. }
复制代码

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)
  1. {
  2.     int flags;
  3.     /* Set the socket nonblocking.
  4.      * Note that fcntl(2) for F_GETFL and F_SETFL can't be
  5.      * interrupted by a signal. */
  6.     if ((flags = fcntl(fd, F_GETFL)) == -1) {
  7.         anetSetError(err, "fcntl(F_GETFL): %s", strerror(errno));
  8.         return ANET_ERR;
  9.     }
  10.     if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
  11.         anetSetError(err, "fcntl(F_SETFL,O_NONBLOCK): %s", strerror(errno));
  12.         return ANET_ERR;
  13.     }
  14.     return ANET_OK;
  15. }
复制代码
我们看到,首先通过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)
  1. {
  2.     int s;
  3.     struct sockaddr_in sa;
  4.     if ((s = anetCreateSocket(err,AF_INET)) == ANET_ERR)
  5.         return ANET_ERR;
  6.     memset(&sa,0,sizeof(sa));
  7.     sa.sin_family = AF_INET;
  8.     sa.sin_port = htons(port);
  9.     sa.sin_addr.s_addr = htonl(INADDR_ANY);
  10.     if (bindaddr && inet_aton(bindaddr, &sa.sin_addr) == 0) {
  11.         anetSetError(err, "invalid bind address");
  12.         close(s);
  13.         return ANET_ERR;
  14.     }
  15.     if (anetListen(err,s,(struct sockaddr*)&sa,sizeof(sa)) == ANET_ERR)
  16.         return ANET_ERR;
  17.     return s;
  18. }
复制代码
TCP服务器的处理流程基本是:创建套接字,初始化地址结构,然后调用anetListene进入监听。监听也分两个步骤,首先把套接字和地址结构绑定的bind,然后开启监听。
服务器创建成功后,返回套接字描述符。然后调用anetGenericAccept开始进入accept状态:static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {
  1.     int fd;
  2.     while(1) {
  3.         fd = accept(s,sa,len);
  4.         if (fd == -1) {
  5.             if (errno == EINTR)
  6.                 continue;
  7.             else {
  8.                 anetSetError(err, "accept: %s", strerror(errno));
  9.                 return ANET_ERR;
  10.             }
  11.         }
  12.         break;
  13.     }
  14.     return fd;
  15. }
复制代码

客户端发起的连接一旦被accept,就可以读写数据了:/* Like read(2) but make sure 'count' is read before to return
  1. * (unless error or EOF condition is encountered) */
  2. int anetRead(int fd, char *buf, int count)
  3. {
  4.     int nread, totlen = 0;
  5.     while(totlen != count) {
  6.         nread = read(fd,buf,count-totlen);
  7.         if (nread == 0) return totlen;
  8.         if (nread == -1) return -1;
  9.         totlen += nread;
  10.         buf += nread;
  11.     }
  12.     return totlen;
  13. }
  14. /* Like write(2) but make sure 'count' is read before to return
  15. * (unless error is encountered) */
  16. int anetWrite(int fd, char *buf, int count)
  17. {
  18.     int nwritten, totlen = 0;
  19.     while(totlen != count) {
  20.         nwritten = write(fd,buf,count-totlen);
  21.         if (nwritten == 0) return totlen;
  22.         if (nwritten == -1) return -1;
  23.         totlen += nwritten;
  24.         buf += nwritten;
  25.     }
  26.     return totlen;
  27. }
复制代码


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


您需要登录后才可以回帖 登录 | 用户注册

本版积分规则

Archiver|手机版|小黑屋|ACE Developer ( 京ICP备06055248号 )

GMT+8, 2024-11-22 02:53 , Processed in 0.015855 second(s), 5 queries , Redis On.

Powered by Discuz! X3.5

© 2001-2023 Discuz! Team.

快速回复 返回顶部 返回列表