unp&chapter6 IO复用

// 每次循环都要重新设置 fd 集合
// O(n) 扫描所有 fd
//简单、跨平台

#include <stdio.h>
#include <sys/select.h>
#include <unistd.h>

int main() {
    fd_set rfds;
    int maxfd = 0; // 最大 fd
    int stdin_fd = STDIN_FILENO;

    while (1) {
        FD_ZERO(&rfds);
        FD_SET(stdin_fd, &rfds);
        if (stdin_fd > maxfd) maxfd = stdin_fd;

        int ret = select(maxfd + 1, &rfds, NULL, NULL, NULL);
        if (ret > 0) {
            if (FD_ISSET(stdin_fd, &rfds)) {
                char buf[128];
                int n = read(stdin_fd, buf, sizeof(buf)-1);
                buf[n] = 0;
                printf("read from stdin: %s", buf);
            }
        }
    }
    return 0;
}

//使用 pollfd 数组
//每次调用仍然遍历数组（O(n)）
//可以方便处理多个 fd，扩展比 select 好
#include <stdio.h>
#include <poll.h>
#include <unistd.h>

int main() {
    struct pollfd fds[1];
    fds[0].fd = STDIN_FILENO;
    fds[0].events = POLLIN;

    while (1) {
        int ret = poll(fds, 1, -1); // 阻塞
        if (ret > 0) {
            if (fds[0].revents & POLLIN) {
                char buf[128];
                int n = read(fds[0].fd, buf, sizeof(buf)-1);
                buf[n] = 0;
                printf("read from stdin: %s", buf);
            }
        }
    }
    return 0;
}

//内核维护活跃 fd 列表
//只处理就绪事件，O(1)
//可以轻松扩展到上万 fd，适合高并发服务器
//支持 LT / ET 模式，可控制触发策略
#include <stdio.h>
#include <sys/epoll.h>
#include <unistd.h>

int main() {
    int epfd = epoll_create1(0);
    struct epoll_event ev;

    ev.events = EPOLLIN;
    ev.data.fd = STDIN_FILENO;
    epoll_ctl(epfd, EPOLL_CTL_ADD, STDIN_FILENO, &ev);

    struct epoll_event events[10];

    while (1) {
        int n = epoll_wait(epfd, events, 10, -1); // 阻塞
        for (int i = 0; i < n; i++) {
            if (events[i].data.fd == STDIN_FILENO) {
                char buf[128];
                int r = read(STDIN_FILENO, buf, sizeof(buf)-1);
                buf[r] = 0;
                printf("read from stdin: %s", buf);
            }
        }
    }
    return 0;
}

struct sigaction sa;
sa.sa_handler = handler;
sa.sa_flags = SA_RESTART;
sigemptyset(&sa.sa_mask);
sigaction(SIGINT, &sa, NULL);

int safe_select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout) {
    int n;
    do {
        n = select(nfds, readfds, writefds, exceptfds, timeout);
    } while (n < 0 && errno == EINTR);
    return n;
}

#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);
        FD_SET(sockfd, &rset);
        maxfdp1 = max(fileno(fp), sockfd) + 1;
        Select(maxfdp1, &rset, NULL, NULL, NULL);

        if (FD_ISSET(sockfd, &rset)) {  /* socket is readable */
            if (Readline(sockfd, recvline, MAXLINE) == 0)
                err_quit("str_cli: server terminated prematurely");
            Fputs(recvline, stdout);
        }

        if (FD_ISSET(fileno(fp), &rset)) {  /* input is readable */
            if (Fgets(sendline, MAXLINE, fp) == NULL)
                return;     /* all done */
            Writen(sockfd, sendline, strlen(sendline));
        }
    }
}

/*
0	SHUT_RD	关闭读方向
1	SHUT_WR	关闭写方向
2	SHUT_RDWR	同时关闭读写
*/
#include <sys/socket.h>
int shutdown(int sockfd, int howto);

#include "unp.h"

void
str_cli(FILE *fp, int sockfd)
{
    int     maxfdp1, stdineof;
    fd_set  rset;
    char    buf[MAXLINE];
    int     n;

    stdineof = 0;                /* 标记标准输入是否已经EOF，用来区分正常/异常结束 */
    FD_ZERO(&rset);
    for ( ; ; ) {
        if (stdineof == 0)       /* 只要stdin未EOF，就监听stdin的可读事件 */
            FD_SET(fileno(fp), &rset);

        FD_SET(sockfd, &rset);   /* 始终监听socket的可读事件 */
        maxfdp1 = max(fileno(fp), sockfd) + 1;
        Select(maxfdp1, &rset, NULL, NULL, NULL);

        /* ---- socket可读事件 ---- */
        if (FD_ISSET(sockfd, &rset)) {
            /* 用read读取原始字节缓冲，不使用fgets避免stdio缓冲干扰select判断 */
            if ( (n = Read(sockfd, buf, MAXLINE)) == 0) {
                /*
                 * socket读到EOF，即对端发送FIN。
                 * 若此时stdin也已EOF(我们已发送FIN)，则为正常关闭；
                 * 否则说明服务器在我们输入没结束前提前终止，属于异常。
                 */
                if (stdineof == 1)
                    return;         /* 正常终止 */
                else
                    err_quit("str_cli: server terminated prematurely");
            }

            /* 将收到的数据写到stdout，同样用write避免缓冲问题 */
            Write(fileno(stdout), buf, n);
        }

        /* ---- stdin可读事件 ---- */
        if (FD_ISSET(fileno(fp), &rset)) {
            /* 用read读stdin数据，不用fgets，避免缓冲影响select */
            if ( (n = Read(fileno(fp), buf, MAXLINE)) == 0) {
                /*
                 * stdin读到EOF(用户Ctrl+D)，表示不再发送数据。
                 * 设置标志，并调用shutdown(SHUT_WR)发送FIN，实现TCP半关闭：
                 *   关闭写方向/保留读方向，继续接收服务器数据。
                 */
                stdineof = 1;                        /* 标记stdin已EOF */
                Shutdown(sockfd, SHUT_WR);           /* 发送FIN(关闭写方向) */
                FD_CLR(fileno(fp), &rset);           /* 不再监听stdin */
                continue;
            }

            /* stdin有数据则写入socket，继续交互 */
            Writen(sockfd, buf, n);
        }
    }
}

/* include fig01 */
#include    "unp.h"

int
main(int argc, char **argv)
{
    int                 i, maxi, maxfd, listenfd, connfd, sockfd;
    int                 nready, client[FD_SETSIZE];
    ssize_t             n;
    fd_set              rset, allset;
    char                buf[MAXLINE];
    socklen_t           clilen;
    struct sockaddr_in  cliaddr, servaddr;

    listenfd = Socket(AF_INET, SOCK_STREAM, 0);  
    // 创建TCP监听socket (IPv4, TCP)

    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family      = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port        = htons(SERV_PORT);
    // 初始化本地地址结构，绑定任意IP+固定端口

    Bind(listenfd, (SA *) &servaddr, sizeof(servaddr));
    // 绑定地址结构到listenfd

    Listen(listenfd, LISTENQ);
    // 开始监听，进入 LISTEN 状态

    maxfd = listenfd;          /* initialize */
    // 当前 select 能看到的最大描述符，初始为 listenfd

    maxi = -1;                 /* index into client[] array */
    // client[] 有效最大下标，初始无客户端

    for (i = 0; i < FD_SETSIZE; i++)
        client[i] = -1;        /* -1 indicates available entry */
    // 初始化 client[] 数组，记录每个客户端socket，不使用 = -1

    FD_ZERO(&allset);
    FD_SET(listenfd, &allset);
    // 把 listenfd 放进select监控集合 allset，代表要监控新连接
/* end fig01 */

/* include fig02 */
    for ( ; ; ) {  // 主事件循环
        rset = allset;      /* structure assignment */
        // select 每次修改 rset，所以每次都先赋值一份
        // rset 是本轮实际监听的集合

        nready = Select(maxfd+1, &rset, NULL, NULL, NULL);
        // 阻塞等待：监听read事件
        // 返回可读的fd数量

        if (FD_ISSET(listenfd, &rset)) {    /* new client connection */
            // 如果 listen socket 可读，表明有新的连接到来

            clilen = sizeof(cliaddr);
            connfd = Accept(listenfd, (SA *) &cliaddr, &clilen);
            // 接受新连接，返回新的连接socket

#ifdef  NOTDEF
            printf("new client: %s, port %d\n",
                    Inet_ntop(AF_INET, &cliaddr.sin_addr, 4, NULL),
                    ntohs(cliaddr.sin_port));
#endif
            // 打印新连接信息（被编译选项禁用）

            for (i = 0; i < FD_SETSIZE; i++)
                if (client[i] < 0) {
                    client[i] = connfd; /* save descriptor */
                    break;
                }
            // 在 client[] 找一个空位保存新客户端fd

            if (i == FD_SETSIZE)
                err_quit("too many clients");
            // 达到同时客户端上限，直接退出

            FD_SET(connfd, &allset);    /* add new descriptor to set */
            // 监控新连接的读事件

            if (connfd > maxfd)
                maxfd = connfd;         /* for select */
            // 更新最大 fd 值

            if (i > maxi)
                maxi = i;               /* max index in client[] array */
            // 更新 client 数组有效范围

            if (--nready <= 0)
                continue;               /* no more readable descriptors */
            // 已处理完本次所有事件
        }

        for (i = 0; i <= maxi; i++) {   /* check all clients for data */
            // 遍历所有客户端，看是否有数据可读

            if ( (sockfd = client[i]) < 0)
                continue;
            // -1 表示槽位未使用，跳过

            if (FD_ISSET(sockfd, &rset)) {
                // 客户端socket有数据可读

                if ( (n = Read(sockfd, buf, MAXLINE)) == 0) {
                        /*4connection closed by client */
                    // Read=0 表示对端正常关闭(EOF)

                    Close(sockfd);
                    FD_CLR(sockfd, &allset);
                    client[i] = -1;
                    // 清理该客户端

                } else
                    Writen(sockfd, buf, n);
                    // 回射：直接把客户端发来的数据原样写回去

                if (--nready <= 0)
                    break;              /* no more readable descriptors */
                // 本轮事件处理完，跳出循环
            }
        }
    }
}
/* end fig02 */

// 每个客户端在同一执行线程/流程中按行处理
while (connection_open) {
    // readline 内部会循环 read()，直到读到 '\n' 或 EOF 才返回
    line = readline(connfd);   // 阻塞，直到遇到 '\n'
    if (line == NULL) break;
    writen(connfd, line);
}

// select/循环 或 每连接独立处理，直接按字节流读写
while (connection_open) {
    n = read(connfd, buf, MAXLINE);  // 读到多少返回多少（不等换行）
    if (n == 0) break;               // 对端关闭
    writen(connfd, buf, n);          // 立即回显已读到的数据
}

sigprocmask(SIG_UNBLOCK, &mask); // 信号不阻塞
select(...);  // 如果此时信号来了，返回 EINTR

sigprocmask(SIG_BLOCK, &mask); // 阻塞信号
pselect(..., &mask);           // 原子解除阻塞 + 等待

时间 --->

情况 A: 信号在 select 前
[ sigprocmask解除 ] --> [ 信号到达并处理 ] --> [ select阻塞 ] --> 阻塞可能永远不返回

情况 B: 信号在 select中
[ sigprocmask解除 ] --> [ select阻塞 ] --> [ 信号到达打断 select ] --> select返回EINTR

#include <poll.h>

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

struct pollfd {
    int   fd;         // 文件描述符
    short events;     // 关注的事件
    short revents;    // 发生的事件，由内核填写
};

#include <stdio.h>
#include <poll.h>
#include <unistd.h>

int main() {
    struct pollfd fds[1];
    fds[0].fd = 0;           // 标准输入
    fds[0].events = POLLIN;  // 关注可读事件

    int ret = poll(fds, 1, 5000); // 等待 5 秒

    if (ret == -1) {
        perror("poll");
    } else if (ret == 0) {
        printf("timeout\n");
    } else {
        if (fds[0].revents & POLLIN) {
            char buf[100];
            int n = read(0, buf, sizeof(buf)-1);
            if (n > 0) {
                buf[n] = '\0';
                printf("input: %s\n", buf);
            }
        }
    }

    return 0;
}

/* include fig01 */
#include	"unp.h"
#include	<limits.h>		/* for FOPEN_MAX */

int
main(int argc, char **argv)
{
	int					i, maxi, listenfd, connfd, sockfd;
	int					nready;
	ssize_t				n;
	char				buf[MAXLINE];
	socklen_t			clilen;
	struct pollfd		client[FOPEN_MAX];
	struct sockaddr_in	cliaddr, servaddr;

	listenfd = Socket(AF_INET, SOCK_STREAM, 0);

	bzero(&servaddr, sizeof(servaddr));
	servaddr.sin_family      = AF_INET;
	servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
	servaddr.sin_port        = htons(SERV_PORT);

	Bind(listenfd, (SA *) &servaddr, sizeof(servaddr));

	Listen(listenfd, LISTENQ);

	client[0].fd = listenfd;
	client[0].events = POLLRDNORM;
	for (i = 1; i < FOPEN_MAX; i++)
		client[i].fd = -1;		/* -1 indicates available entry */
	maxi = 0;					/* max index into client[] array */
/* end fig01 */

/* include fig02 */
	for ( ; ; ) {
		nready = Poll(client, maxi+1, INFTIM);

		if (client[0].revents & POLLRDNORM) {	/* new client connection */
			clilen = sizeof(cliaddr);
			connfd = Accept(listenfd, (SA *) &cliaddr, &clilen);
#ifdef	NOTDEF
			printf("new client: %s\n", Sock_ntop((SA *) &cliaddr, clilen));
#endif

			for (i = 1; i < FOPEN_MAX; i++)
				if (client[i].fd < 0) {
					client[i].fd = connfd;	/* save descriptor */
					break;
				}
			if (i == FOPEN_MAX)
				err_quit("too many clients");

			client[i].events = POLLRDNORM;
			if (i > maxi)
				maxi = i;				/* max index in client[] array */

			if (--nready <= 0)
				continue;				/* no more readable descriptors */
		}

		for (i = 1; i <= maxi; i++) {	/* check all clients for data */
			if ( (sockfd = client[i].fd) < 0)
				continue;
			if (client[i].revents & (POLLRDNORM | POLLERR)) {
				if ( (n = read(sockfd, buf, MAXLINE)) < 0) {
					if (errno == ECONNRESET) {
							/*4connection reset by client */
#ifdef	NOTDEF
						printf("client[%d] aborted connection\n", i);
#endif
						Close(sockfd);
						client[i].fd = -1;
					} else
						err_sys("read error");
				} else if (n == 0) {
						/*4connection closed by client */
#ifdef	NOTDEF
					printf("client[%d] closed connection\n", i);
#endif
					Close(sockfd);
					client[i].fd = -1;
				} else
					Writen(sockfd, buf, n);

				if (--nready <= 0)
					break;				/* no more readable descriptors */
			}
		}
	}
}
/* end fig02 */