[原]Redis源代码分析之四：Unix底层网络通信
还是先介绍接口：int anetTcpConnect(char *err, char *addr, int port);int anetTcpNonBlockConnect(char *err, char *addr, int port);
int anetUnixConnect(char *err, char *path);
int anetUnixNonBlockConnect(char *err, char *path);
int anetRead(int fd, char *buf, int count);
int anetResolve(char *err, char *host, char *ipbuf);
int anetTcpServer(char *err, int port, char *bindaddr);
int anetUnixServer(char *err, char *path, mode_t perm);
int anetTcpAccept(char *err, int serversock, char *ip, int *port);
int anetUnixAccept(char *err, int serversock);
int anetWrite(int fd, char *buf, int count);
int anetNonBlock(char *err, int fd);
int anetTcpNoDelay(char *err, int fd);
int anetTcpKeepAlive(char *err, int fd);
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)
{
int s;
struct sockaddr_in sa;
if ((s = anetCreateSocket(err,AF_INET)) == ANET_ERR)
return ANET_ERR;
sa.sin_family = AF_INET;
sa.sin_port = htons(port);
if (inet_aton(addr, &sa.sin_addr) == 0) {
struct hostent *he;
he = gethostbyname(addr);
if (he == NULL) {
anetSetError(err, "can't resolve: %s", addr);
close(s);
return ANET_ERR;
}
memcpy(&sa.sin_addr, he->h_addr, sizeof(struct in_addr));
}
if (flags & ANET_CONNECT_NONBLOCK) {
if (anetNonBlock(err,s) != ANET_OK)
return ANET_ERR;
}
if (connect(s, (struct sockaddr*)&sa, sizeof(sa)) == -1) {
if (errno == EINPROGRESS &&
flags & ANET_CONNECT_NONBLOCK)
return s;
anetSetError(err, "connect: %s", strerror(errno));
close(s);
return ANET_ERR;
}
return s;
}
首先调用anetCreateSocket函数创建套接字：static int anetCreateSocket(char *err, int domain) {
int s, on = 1;
if ((s = socket(domain, SOCK_STREAM, 0)) == -1) {
anetSetError(err, "creating socket: %s", strerror(errno));
return ANET_ERR;
}
/* Make sure connection-intensive things like the redis benckmark
* will be able to close/open sockets a zillion of times */
if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)) == -1) {
anetSetError(err, "setsockopt SO_REUSEADDR: %s", strerror(errno));
return ANET_ERR;
}
return s;
}
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)
{
int flags;
/* Set the socket nonblocking.
* Note that fcntl(2) for F_GETFL and F_SETFL can't be
* interrupted by a signal. */
if ((flags = fcntl(fd, F_GETFL)) == -1) {
anetSetError(err, "fcntl(F_GETFL): %s", strerror(errno));
return ANET_ERR;
}
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
anetSetError(err, "fcntl(F_SETFL,O_NONBLOCK): %s", strerror(errno));
return ANET_ERR;
}
return ANET_OK;
}
我们看到，首先通过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)
{
int s;
struct sockaddr_in sa;
if ((s = anetCreateSocket(err,AF_INET)) == ANET_ERR)
return ANET_ERR;
memset(&sa,0,sizeof(sa));
sa.sin_family = AF_INET;
sa.sin_port = htons(port);
sa.sin_addr.s_addr = htonl(INADDR_ANY);
if (bindaddr && inet_aton(bindaddr, &sa.sin_addr) == 0) {
anetSetError(err, "invalid bind address");
close(s);
return ANET_ERR;
}
if (anetListen(err,s,(struct sockaddr*)&sa,sizeof(sa)) == ANET_ERR)
return ANET_ERR;
return s;
}
TCP服务器的处理流程基本是：创建套接字，初始化地址结构，然后调用anetListene进入监听。监听也分两个步骤，首先把套接字和地址结构绑定的bind，然后开启监听。
服务器创建成功后，返回套接字描述符。然后调用anetGenericAccept开始进入accept状态：static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {
int fd;
while(1) {
fd = accept(s,sa,len);
if (fd == -1) {
if (errno == EINTR)
continue;
else {
anetSetError(err, "accept: %s", strerror(errno));
return ANET_ERR;
}
}
break;
}
return fd;
}
客户端发起的连接一旦被accept，就可以读写数据了：/* Like read(2) but make sure 'count' is read before to return
* (unless error or EOF condition is encountered) */
int anetRead(int fd, char *buf, int count)
{
int nread, totlen = 0;
while(totlen != count) {
nread = read(fd,buf,count-totlen);
if (nread == 0) return totlen;
if (nread == -1) return -1;
totlen += nread;
buf += nread;
}
return totlen;
}
/* Like write(2) but make sure 'count' is read before to return
* (unless error is encountered) */
int anetWrite(int fd, char *buf, int count)
{
int nwritten, totlen = 0;
while(totlen != count) {
nwritten = write(fd,buf,count-totlen);
if (nwritten == 0) return totlen;
if (nwritten == -1) return -1;
totlen += nwritten;
buf += nwritten;
}
return totlen;
}
作者：Aegeaner 发表于2012-2-4 14:29:33 原文链接
页:
[1]