一.前言
从上个世纪到现在,工程师们在优化服务器性能的过程中,提出了各种不同的io模型,比如非阻塞io,io复用,信号驱动式io,异步io。具体io模型在不同平台上的实现也不一样,比如io复用在bsd上可以由kqueue实现,在solaris系统上可以由/dev/poll实现。为了实现系统的可移植性,POSIX 确保 select和poll在 unix-like系统上得到广泛的支持。
在上个世纪,Dan Kegel 提出了C10K的设想,现在C10K 已经不是什么问题,比如nginx就可以做到百万级别的qps。于是又有人提出来了C10M的设想,Robert David Graham 从unix的最初设计初衷给出了自己的解决方案。
二.常见io模型
1.阻塞io
常见的read系统调用,是最常见的阻塞io:
2.非阻塞式io
非阻塞io的典型使用方式如下,设置非阻塞标志,并且常与io复用一起使用,使用起来比较复杂。
val = Fcntl(sockfd, F_GETFL, 0);
Fcntl(sockfd, F_SETFL, val | O_NONBLOCK); /* O_NONBLOCK 标志非阻塞 */
3.io 复用 (select/poll)
io复用在处理数量庞大的fd时非常有效,我们以select为例,select的核心api是select函数:
int select(int nfds, fd_set *_Nullable restrict readfds,
fd_set *_Nullable restrict writefds,
fd_set *_Nullable restrict exceptfds,
struct timeva l *_Nullable restrict timeout);
看一个例子:
#include "unp.h"
void
str_cli(FILE *fp, int sockfd)
{
int maxfdp1;
fd_set rset;
char sendline[MAXLINE], recvline[MAXLINE];
FD_ZERO(&rset);
for ( ; ; ) {
FD_SET(fileno(fp), &rset); /* 设置要监听的socket fd */
FD_SET(sockfd, &rset); /* 设置要监听的file fd */
maxfdp1 = max(fileno(fp), sockfd) + 1;
Select(maxfdp1, &rset, NULL, NULL, NULL); /* select 调用 */
if (FD_ISSET(sockfd, &rset)) { /* socket 可读 */
if (Readline(sockfd, recvline, MAXLINE) == 0)
err_quit("str_cli: server terminated prematurely");
Fputs(recvline, stdout);
}
if (FD_ISSET(fileno(fp), &rset)) { /* input 可读 */
if (Fgets(sendline, MAXLINE, fp) == NULL)
return; /* all done */
Writen(sockfd, sendline, strlen(sendline));
}
}
}
4.信号驱动式io
但凡涉及到信号的程序都比较复杂。要使用信号驱动式io,先开启socket的信号驱动式io功能,并通过sigaction 系统调用安装一个信号处理函数:
void
dg_echo(int sockfd_arg, SA *pcliaddr, socklen_t clilen_arg)
{
int i;
const int on = 1;
sigset_t zeromask, newmask, oldmask;
sockfd = sockfd_arg;
clilen = clilen_arg;
for (i = 0; i < QSIZE; i++) { /* init queue of buffers */
dg[i].dg_data = Malloc(MAXDG);
dg[i].dg_sa = Malloc(clilen);
dg[i].dg_salen = clilen;
}
iget = iput = nqueue = 0;
Signal(SIGHUP, sig_hup); /* 安装信号处理函数 */
Signal(SIGIO, sig_io);
Fcntl(sockfd, F_SETOWN, getpid()); /* 设置属主 */
Ioctl(sockfd, FIOASYNC, &on); /* 开启信号驱动式io */
Ioctl(sockfd, FIONBIO, &on); /* non-bloking */
Sigemptyset(&zeromask); /* init three signal sets */
Sigemptyset(&oldmask);
Sigemptyset(&newmask);
Sigaddset(&newmask, SIGIO); /* signal we want to block */
Sigprocmask(SIG_BLOCK, &newmask, &oldmask);
for ( ; ; ) {
while (nqueue == 0)
sigsuspend(&zeromask); /* wait for datagram to process */
/* 4unblock SIGIO */
Sigprocmask(SIG_SETMASK, &oldmask, NULL);
Sendto(sockfd, dg[iget].dg_data, dg[iget].dg_len, 0,
dg[iget].dg_sa, dg[iget].dg_salen);
if (++iget >= QSIZE)
iget = 0;
/* 4block SIGIO */
Sigprocmask(SIG_BLOCK, &newmask, &oldmask);
nqueue--;
}
}
5.异步io
我们来看一个aio的例子(由于aio的例子过于复杂,我们这里只截取部分关键代码):
for (i = 0; i < NBUF; i++) {
switch (bufs[i].op) {
case UNUSED:
/*
* Read from the input file if more data
* remains unread.
*/
if (off < sbuf.st_size) {
bufs[i].op = READ_PENDING;
bufs[i].aiocb.aio_fildes = ifd;
bufs[i].aiocb.aio_offset = off;
off += BSZ;
if (off >= sbuf.st_size)
bufs[i].last = 1;
bufs[