|
还是先介绍接口: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 原文链接
|
|