首页 > 代码库 > Socket编程实践(12) --select实现超时I/O API[比较难于理解/代码较多]
Socket编程实践(12) --select实现超时I/O API[比较难于理解/代码较多]
read_timeout函数封装
//读超时函数,但不包含读操作
int read_timeout(int fd, long waitSec)
{
int returnValue = http://www.mamicode.com/0;>
write_timeout函数封装
//写超时函数,但不包含读操作
int write_timeout(int fd, long waitSec)
{
int returnValue = http://www.mamicode.com/0;>
accept_timeout函数封装
/**超时accept函数
成功:返回已经建立链接的文件描述符
失败:返回-1,errno=ETIMEDOUT
*/
int accept_timeout(int fd, struct sockaddr_in *addr, long waitSec)
{
int returnValue = http://www.mamicode.com/0;>
connect_timeout函数封装
1.我们为什么需要这个函数????
TCP/IP在客户端连接服务器时,如果发生异常,connect(如果是在默认阻塞的情况下),返回的是1.5*RTT(相当于客户端阻塞了这么长的时间,客户需要等待这么长的时间,显然这样的客户端用户体验并不好);会造成严重的软件质量下降.
2.怎样实现?
1)sockfd首先变成非阻塞的;然后试着进行connect,如果网络状况良好,则立刻建立链接并返回,如果网络状况不好,则链接不会马上建立,这时需要我们的参与:调用select,设置等待时间,通过select管理者去监控sockfd,一旦能够建立链接,则马上返回,然后建立链接,这样就会大大提高我们的软件质量.
2)需要注意一点:select机制监控到sockfd可写(也就是可以建立链接时),并不代表调用accept就一定能够成功(造成sockfd可写有两种情况:1.真正的链接可以建立起来了;2.建立链接的过程中发生错误,然后错误会回写错误信息,造成sockfd可写);
通过调用sockoptret做一个容错即可!
代码实现:
//设置文件描述符fd为非阻塞模式
int setUnBlock(int fd)
{
int flags = fcntl(fd,F_GETFL);
flags |= O_NONBLOCK;
if (fcntl(fd,F_SETFL,flags) == -1)
{
err_exit("fcntl O_NONBLOCK error");
}
return 0;
}
//设置文件描述符fd为阻塞模式
int setBlock(int fd)
{
int flags = fcntl(fd,F_GETFL);
flags &= ~O_NONBLOCK;
if (fcntl(fd,F_SETFL,flags) == -1)
{
err_exit("fcntl BLOCK error");
}
return 0;
}
/**说明:
fd:套接字
addr:要连接的对方的地址
waitSec:等待超时的秒数,如果为0,则为正常模式
返回值:成功(没有超时)返回0,失败/超时返回-1,errno=ETIMEDOUT
*/
int connect_timeout(int fd, struct sockaddr_in *addr, long waitSec)
{
socklen_t addrLen = sizeof(struct sockaddr_in);
if (waitSec > 0)
{
//设置为非阻塞模式
setUnBlock(fd);
}
//首先尝试着进行链接
int returnValue = http://www.mamicode.com/connect(fd,(struct sockaddr *)addr,addrLen);>
完整的server端调用代码:
//server端完整代码及解析
#include "commen.h"
#include "timeout.h"
int main()
{
signal(SIGCHLD,onCatchSIGCHLD);
int serverSockfd = mkATCPServer(9001);
long int waitSec = 10; //超时时间为10秒
struct sockaddr_in peerAddr;
while (true)
{
//接受链接,只等待100秒,如果链接不来,我就停止服务(这儿只是为了演示,其实这是不合理的)
int peerSockfd = accept_timeout(serverSockfd,&peerAddr,100);
if (peerSockfd == -1)
{
err_exit("main: accept error");
}
//打印客户信息
cout << "Client:" << endl;
cout << "\tsin_port: " << ntohs(peerAddr.sin_port) << endl;
cout << "\tsin_addr: " << inet_ntoa(peerAddr.sin_addr) << endl;
cout << "\tsocket: " << peerSockfd << endl;
//每有一个客户端连接进来,就fork一个子进程,
//相应的业务处理由子进程完成,父进程继续监听
pid_t pid = fork();
if (pid == -1)
{
close(serverSockfd);
close(peerSockfd);
err_exit("fork error");
}
else if (pid == 0) //子进程,处理业务
{
close(serverSockfd); //子进程关闭监听套接字,因为子进程不负责监听任务
char recvBuf[BUFSIZ];
ssize_t readCount = 0;
while (true)
{
/**设定读等待的秒数:
1.如果在这段时间中有事件(socket 可读)到来,则返回
2.如果一直都没有事件到来,则select会一直等待,直到超时返回(-1)
*/
int ret = read_timeout(peerSockfd,waitSec);
if (ret == 0)
{
memset(recvBuf,0,sizeof(recvBuf));
//此时再进行读取socket就一定不会阻塞了O(∩_∩)O~
if ((readCount = read(peerSockfd,recvBuf,sizeof(recvBuf))) == -1)
{
err_exit("readn error");
}
else if (readCount == 0)
{
peerClosePrint("client connect closed");
}
}
else if (ret == -1 && errno == ETIMEDOUT) //超时
{
err_exit("read time out");
}
/**设定写等待的秒数:
1.如果在这段时间中有事件(socket 可写)到来,则返回
2.如果一直都没有事件到来,则select会一直等待,直到超时返回(-1)
*/
ret = write_timeout(peerSockfd,waitSec);
if (ret == 0)
{
//将整体报文回写回客户端
if (writen(peerSockfd,recvBuf,strlen(recvBuf)) == -1)
{
err_exit("writen error");
}
}
else if (ret == -1 && errno == ETIMEDOUT) //超时
{
err_exit("write time out");
}
recvBuf[readCount] = 0;
//写至终端
fputs(recvBuf,stdout);
}
}
else if (pid > 0) //父进程
{
close(peerSockfd);
}
}
close(serverSockfd);
return 0;
}
完整的client端调用代码:
#include "commen.h"
int main()
{
int sockfd = mkATCPClient(9001,"127.0.0.1");
char sendBuf[BUFSIZ];
char recvBuf[BUFSIZ];
while (true)
{
int fdCount = sockfd > STDIN_FILENO ? sockfd+1 : STDIN_FILENO+1;
fd_set rdset;
FD_ZERO(&rdset);
FD_SET(sockfd,&rdset);
FD_SET(STDIN_FILENO,&rdset);
int nReady = select(fdCount,&rdset,NULL,NULL,NULL);
if (nReady == -1)
{
err_exit("select error");
}
//server端有数据可读
if (FD_ISSET(sockfd,&rdset))
{
//从socket中读取数据
int readCount = read(sockfd,recvBuf,sizeof(recvBuf));
if ( readCount == -1)
{
err_exit("read socket error");
}
else if (readCount == 0) //如果对端结束链接
{
peerClosePrint();
}
fputs(recvBuf,stdout);
memset(recvBuf,0,sizeof(recvBuf));
}
//标准输入上有数据可读:从键盘读取数据 -> 发送至socket
if (FD_ISSET(STDIN_FILENO,&rdset));
{
if(fgets(sendBuf,sizeof(sendBuf),stdin) != NULL)
{
if (write(sockfd,sendBuf,strlen(sendBuf)) == -1) //发送到socket上
{
err_exit("write error");
}
memset(sendBuf,0,sizeof(sendBuf));
}
}
}
close(sockfd);
return 0;
}
结果演示:
附-commen.h
#ifndef COMMEN_H_INCLUDED
#define COMMEN_H_INCLUDED
#include <unistd.h>
#include <signal.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/msg.h>
#include <sys/sem.h>
#include <sys/socket.h>
#include <sys/select.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <iostream>
using namespace std;
#include "timeout.h"
void peerClosePrint(std::string str = "peer connect closed")
{
cout << str << endl;
_exit(0);
}
//return a socket that have start listened.
int mkATCPServer(int serverPort, int backlog = SOMAXCONN)
{
int sockfd = socket(AF_INET,SOCK_STREAM,0);
if (sockfd == -1)
{
err_exit("socket error");
}
//add address reused
int on = 1;
if (setsockopt(sockfd,SOL_SOCKET,SO_REUSEADDR,&on,sizeof(on)) == -1)
{
err_exit("setsockopt SO_REUSEADDR error");
}
//band a local address and port
struct sockaddr_in serverAddr;
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = htons(serverPort);
serverAddr.sin_addr.s_addr = INADDR_ANY; //band an any IP address
if (bind(sockfd,(struct sockaddr *)&serverAddr,sizeof(serverAddr)) == -1)
{
err_exit("bind error");
}
//start to listen.
if (listen(sockfd,backlog) == -1)
{
err_exit("listen error");
}
return sockfd;
}
//return a socket that have connected to server.
int mkATCPClient(int serverPort, string serverIPAddr)
{
//first. create a socket
int sockfd = socket(AF_INET,SOCK_STREAM,0);
if (sockfd == -1)
{
err_exit("socket error");
}
//second. connect to a server
struct sockaddr_in serverAddr;
serverAddr.sin_family = AF_INET;
serverAddr.sin_port = htons(serverPort);
serverAddr.sin_addr.s_addr = inet_addr(serverIPAddr.c_str());
if (connect_timeout(sockfd,&serverAddr,10) == -1)
{
err_exit("In mkATCPClient -> connect error");
}
return sockfd;
}
void onCatchSIGCHLD(int signalNumber)
{
int ret = 0;
while ((ret = waitpid(-1,NULL,WNOHANG) != -1))
;
}
ssize_t readn(int fd,void *buf,size_t count)
{
size_t nLeft = count;
ssize_t nRead = 0;
char *ptr = static_cast<char *>(buf);
while (nLeft > 0)
{
if ((nRead = read(fd,ptr,nLeft)) < 0)
{
//一点东西都没读
if (nLeft == count)
{
return -1; //error
}
else
{
break; //error, return amount read so far
}
}
else if (nRead == 0)
{
break; //EOF
}
nLeft -= nRead;
ptr += nRead;
}
return count - nLeft;
}
ssize_t writen(int fd, const void *buf, size_t count)
{
size_t nLeft = count;
ssize_t nWritten;
const char *ptr = static_cast<const char *>(buf);
while (nLeft > 0)
{
if ((nWritten = write(fd,ptr,nLeft)) < 0)
{
//一点东西都没写
if (nLeft == count)
{
return -1; //error
}
else
{
break; //error, return amount write so far
}
}
else if (nWritten == 0)
{
break; //EOF
}
nLeft -= nWritten;
ptr += nWritten;
}
return count - nWritten;
}
//只是查看一下网络中的数据,并不是将之真正取走:MSG_PEEK
ssize_t recv_peek(int fd, void *buf, size_t count)
{
int nRead = 0;
//如果读取网络数据出错,则继续读取
while ((nRead = recv(fd,buf,count,MSG_PEEK)) == -1);
return nRead;
}
ssize_t readline(int fd, void *buf, size_t maxline)
{
char *pBuf = (char *)buf;
int nLeft = maxline;
while (true)
{
//查看缓冲区中的数据,并不真正取走
int nTestRead = recv_peek(fd,pBuf,nLeft);
//检测这次读来的数据中是否包含‘\n‘;
//如果有,则将之全部读取出来
for (int i = 0; i < nTestRead; ++i)
{
if (pBuf[i] == ‘\n‘)
{
//真正的从缓冲区中将数据取走
if (readn(fd,pBuf,i+1) != i+1)
{
err_exit("readn error");
}
else
{
return i + 1;
}
}
}
//如果这次读的缓冲区中没有‘\n‘
//如果读超了:读道德数目大于一行最大数,则做异常处理
if (nTestRead > nLeft)
{
exit(EXIT_FAILURE);
}
nLeft -= nTestRead; //若缓冲区没有‘\n‘,则将剩余的数据读走
if (readn(fd,pBuf,nTestRead) != nTestRead)
{
exit(EXIT_FAILURE);
}
pBuf += nTestRead;
}
return -1;
}
#endif // COMMEN_H_INCLUDED
附-timeout.h
#ifndef TIMEOUT_H_INCLUDED
#define TIMEOUT_H_INCLUDED
static void err_exit(std::string str)
{
perror(str.c_str());
exit(EXIT_FAILURE);
}
/*
struct timeval
{
long tv_sec; // seconds //
long tv_usec; // microseconds //
};
*/
//读超时函数,但不包含读操作
int read_timeout(int fd, long waitSec)
{
int returnValue = http://www.mamicode.com/0;>
附-Makefile
CC = g++
CPPFLAGS = -Wall -g -pthread
BIN = server client
SOURCES = $(BIN.=.cpp)
.PHONY: clean all
all: $(BIN)
$(BIN): $(SOURCES)
clean:
-rm -rf $(BIN) bin/ obj/ core
附-RTT(Round-Trip Time):
往返时延。在计算机网络中它是一个重要的性能指标，表示从发送端发送数据开始，到发送端收到来自接收端的确认（接收端收到数据后便立即发送确认），总共经历的时延。
往返延时(RTT)由三个部分决定：即链路的传播时间、末端系统的处理时间以及路由器的缓存中的排队和处理时间。其中，前面两个部分的值作为一个TCP连接相对固定，路由器的缓存中的排队和处理时间会随着整个网络拥塞程度的变化而变化。所以RTT的变化在一定程度上反映了网络拥塞程度的变化。简单来说就是发送方从发送数据开始，到收到来自接受方的确认信息所经历的时间。
Socket编程实践(12) --select实现超时I/O API[比较难于理解/代码较多]
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。