zoukankan      html  css  js  c++  java
  • 非阻塞式I/O

      套接字的默认状态是阻塞的。这就意味着当发出一个不能立即完成的套接字调用时,其进程将被投入睡眠,等待相应的操作完成。可能阻塞的套接字调用可分为以下4类:

      (1)输入操作,包括read,readv,recv,recvfrom和recvmsg共5个函数。如果某个进程对一个阻塞的TCP套接字(默认设置)调用这些输入函数之一,而且该套接字的接收缓冲区中没有数据可读,该进程将被投入睡眠,直到有一些数据到达。既然TCP是字节流协议,该进程的唤醒就是只要有一些数据到达,这些数据既可能是单个字节,也可能是一个完整的TCP分节中的数据,所以可以指定MSG_WAITALL标志。

      既然UDP是数据报协议,如果一个阻塞的UDP套接字的接收缓冲区为空,对它调用输入函数的进程将被投入睡眠,直到有UDP数据报到达。

      对于非阻塞的套接字,如果输入操作不能被满足(对于TCP套接字即至少有一个字节的数据报可读,对于UDP套接字即有一个完整的数据报可读),相应调用将立即返回一个EWOULDBLOCK错误。

      (2)输出操作,包括write,writev,send,sento和sendmsg共5个函数。对于一个TCP套接字,内核将从应用进程的缓冲区到该套接字的发送缓冲区复制数据。对于阻塞的套接字,如果其发送缓冲区中没有空间,进程将被投入睡眠,直到有空间为止。

      对于一个非阻塞的TCP套接字,如果其发送缓冲区中根本没有空间,输出函数调用将立即返回一个EWOULDBLOCK错误。如果其发送缓冲区中有一些空间,返回值将是内核能够复制到该缓冲区中的字节数。

      UDP套接字不存在真正的发送缓冲区。内核只是复制应用进程数据并把它沿协议栈向下传送,渐次冠以UDP首部和IP首部。因此对一个阻塞的UDP套接字(默认设置),输出函数调用将不会因与TCP套接字一样的原因而阻塞,不过有可能会因为其他的原因而阻塞。

      (3)接受外来连接,即accept函数。如果对一个阻塞的套接字调用accept函数,并且尚无新的连接到达,调用进程将被投入睡眠。如果对一个非阻塞的套接字调用accept函数,并且尚无新的连接到达,accept调用将立即返回一个EWOULDBLOCK错误。

      (4)发起外出连接,即用于TCP的connect函数(connect也可用于UDP,不过它不能是一个“真正”的连接建立起来,他只是使内核保存对端的IP地址和端口号)。TCP连接的建立涉及一个三次握手过程,而且connect函数一直等到客户收到自己的SYN的ACK为止才返回。这意味着TCP的每个connect总是阻塞其调用进程至少一个到服务器的RTT时间。

      如果对一个非阻塞的TCP套接字调用connect,并且连接不能立即建立,那么连接的建立能照样发起(譬如送出TCP三次握手的第一个分组),不过会返回一个EINPROGRESS错误。注意这个错误不同于上述三个情形中的返回的错误。

    非阻塞读和写(str_cli函数)

      如果client调用write把内容发送给服务器,此时发送缓冲区已满,write则会阻塞,在此阻塞期间,可能有来自套接字缓冲区的数据可读;类似,如果从套接字缓冲区有一行文本输入可读,那么一旦标准输入比网络还要慢,那么进程可能阻塞于后续的write调用。

      以下程序维护两个缓冲区:to容纳从标准输入到服务器去的数据,fr容纳自服务器到标准输出的数据。

    void str_cli(FILE *fp, int sockfd)
    {
        int            maxfdp1, val, stdineof;
        ssize_t        n, nwritten;
        fd_set        rset, wset;
        char        to[MAXLINE], fr[MAXLINE];
        char        *toiptr, *tooptr, *friptr, *froptr;
     
        val = Fcntl(sockfd, F_GETFL, 0);
        Fcntl(sockfd, F_SETFL, val | O_NONBLOCK);
     
        val = Fcntl(STDIN_FILENO, F_GETFL, 0);
        Fcntl(STDIN_FILENO, F_SETFL, val | O_NONBLOCK);
     
        val = Fcntl(STDOUT_FILENO, F_GETFL, 0);
        Fcntl(STDOUT_FILENO, F_SETFL, val | O_NONBLOCK);
     
        toiptr = tooptr = to;    /* initialize buffer pointers */
        friptr = froptr = fr;
        stdineof = 0;
     
        maxfdp1 = max(max(STDIN_FILENO, STDOUT_FILENO), sockfd) + 1;
        for ( ; ; ) 
        {
            FD_ZERO(&rset);
            FD_ZERO(&wset);
            if (stdineof == 0 && toiptr < &to[MAXLINE])
                FD_SET(STDIN_FILENO, &rset);    /* read from stdin */
            if (friptr < &fr[MAXLINE])
                FD_SET(sockfd, &rset);            /* read from socket */
            if (tooptr != toiptr)
                FD_SET(sockfd, &wset);            /* data to write to socket */
            if (froptr != friptr)
                FD_SET(STDOUT_FILENO, &wset);    /* data to write to stdout */
     
            Select(maxfdp1, &rset, &wset, NULL, NULL);
    
            if (FD_ISSET(STDIN_FILENO, &rset))
            {
                if ( (n = read(STDIN_FILENO, toiptr, &to[MAXLINE] - toiptr)) < 0) 
                {
                    if (errno != EWOULDBLOCK)
                        err_sys("read error on stdin");
                }
                else if (n == 0)
                {
                    fprintf(stderr, "%s: EOF on stdin
    ", gf_time());
                    stdineof = 1;            /* all done with stdin */
                    if (tooptr == toiptr)//如果缓冲区不再有数据发送,就调用shutdown发送FIN到服务器;如果仍有数据发送,就推迟到缓冲区中数据写到套接字之后
                        Shutdown(sockfd, SHUT_WR);/* send FIN */
                } 
                else
                {
                    fprintf(stderr, "%s: read %d bytes from stdin
    ", gf_time(), n);
                    toiptr += n;            /* # just read */
                    FD_SET(sockfd, &wset);    /* try and write to socket below */
                }
            }
            if (FD_ISSET(sockfd, &rset)) 
            {
                if ( (n = read(sockfd, friptr, &fr[MAXLINE] - friptr)) < 0)
                {
                    if (errno != EWOULDBLOCK)
                        err_sys("read error on socket");
                }
                else if (n == 0)
                {
                    fprintf(stderr, "%s: EOF on socket
    ", gf_time());
                    if (stdineof)
                        return;        /*如果在标准输入上遇到EOF,则为正常的normal termination */
                    else
                        err_quit("str_cli: server terminated prematurely");//否则来自服务器的EOF并为预期
                } 
                else
                {
                    fprintf(stderr, "%s: read %d bytes from socket
    ",gf_time(), n);
                    friptr += n;        /* # just read */
                    FD_SET(STDOUT_FILENO, &wset);    /* try and write below */
                }
            }
            if (FD_ISSET(STDOUT_FILENO, &wset) && ( (n = friptr - froptr) > 0)) 
            {
                if ( (nwritten = write(STDOUT_FILENO, froptr, n)) < 0) 
                {
                    if (errno != EWOULDBLOCK)
                        err_sys("write error to stdout");
                }
                else 
                {
                    fprintf(stderr, "%s: wrote %d bytes to stdout
    ",gf_time(), nwritten);
                    froptr += nwritten;        /* # just written */
                    if (froptr == friptr)
                        froptr = friptr = fr;    /* back to beginning of buffer */
                }
            }
            if (FD_ISSET(sockfd, &wset) && ( (n = toiptr - tooptr) > 0)) 
            {
                if ( (nwritten = write(sockfd, tooptr, n)) < 0) 
                {
                    if (errno != EWOULDBLOCK)
                        err_sys("write error to socket");
                } 
                else
                {
                    fprintf(stderr, "%s: wrote %d bytes to socket
    ",gf_time(), nwritten);
                    tooptr += nwritten;    /* # just written */
                    if (tooptr == toiptr) 
                    {
                        toiptr = tooptr = to;    /* back to beginning of buffer */
                        if (stdineof)
                            Shutdown(sockfd, SHUT_WR);    /* send FIN */
                    }
                }
            }
        }
    }

      str_cli的简单版本

      每当需要使用非阻塞式I/O时,更简单的办法是把应用程序任务划分为多个进程(fork)或多个线程。

      这个函数一开始就调用fork把当前进程划分成一个父进程和一个子进程。子进程把来自服务器的文本行复制到标准输出,父进程把来自标准输入的文本行复制到服务器。

    void str_cli(FILE *fp, int sockfd)
    {
        pid_t    pid;
        char    sendline[MAXLINE], recvline[MAXLINE];
     
        if ( (pid = Fork()) == 0) {        /* child: server -> stdout */
            while (Readline(sockfd, recvline, MAXLINE) > 0)
                Fputs(recvline, stdout);
     
            kill(getppid(), SIGTERM);    /* in case parent still running */
            exit(0);
        }
     
            /* parent: stdin -> server */
        while (Fgets(sendline, MAXLINE, fp) != NULL)
            Writen(sockfd, sendline, strlen(sendline));
     
        Shutdown(sockfd, SHUT_WR);    /* EOF on stdin, send FIN */
        pause();
        return;
    }

         我们知道TCP连接是全双工的,而且父子进程共享同一个套接字:父进程往该套接字中写,子进程从该套接字中读。尽管套接字只有一个,其接收缓冲区和发送缓冲区也分别只有一个,然而这个套接字却有两个描述符在引用它:一个在父进程中,另一个在子进程中。

         我们同样需要考虑进程终止序列。正常的终止序列从标准输入上遇到EOF之时开始发生。父进程读入来自标准输入的EOF后调用shutdown发送FIN。但当这发生之后,子进程需继续从服务器到标准输出执行数据复制,直到在套接字上读到EOF。

         服务器进程过早终止也有可能发生。要是发生这种情况,子进程将在套接字上读到EOF。这样的子进程必须告知父进程停止从标准输入到套接字复制数据。子进程向父进程发送一个SIGTERM信号,以防止父进程仍在运行。如此处理的另一个手段是子进程无为的终止,使得父进程(如果仍在运行的话)捕获一个SIGCHLD信号。

         父进程完成数据复制后调用pause让自己进入睡眠状态,直到捕获一个信号(子进程来的SIGTERM信号),尽管它不主动捕获任何信号。SIGTERM信号的默认行为是终止进程。

     非阻塞connect

      当在一个非阻塞的TCP套接字上调用connect时,connect将立即返回一个EINPROGRESS错误,不过已经发起的TCP三路握手继续进行。非阻塞的connect有三个用途

      1.我们可以把三路握手叠加在其他处理上。完成一个connect要花一个RTT时间,其波动范围很大,从局域网上的几个毫秒到几百个毫秒甚至是广域网的几秒。这段时间可以处理其他工作。

      2.可以使用这个技术同时建立多个连接。

      3.使用select等待连接的建立,我们可以给select指定一个时间限制,使得我们能够缩短connect的超时。许多实现有着从75毫秒到数分钟的connect超时时间。应用程序有时想要一个更短的超时时间,可以使用非阻塞的connect.

      怎么使用connect:

    1. 先将套接字描述符设置为非阻塞,然后按照常规网络编程,直到调用connect时,connect将立即返回一个EINPROGRESS错误(ps: 已经发起的三路握手继续执行)。我们接着使用I/O复用(比如 select)检测这个连接或成功或失败的已建立条件,需要注意的处理细节。
    2. 尽管套接字是非阻塞的,如果连接的服务器在同一主机上,我们调用connect时,连接通常立刻建立。
    3. 关于select和非阻塞connect的以下两个规则:当连接成功建立时,描述符变为可写;当连接建立遇到错误时,描述符变为既可读又可写(TCP套接字可写的条件是发送缓冲区中有可用空间)。
    int connect_nonb(int sockfd, const SA *saptr, socklen_t salen, int nsec)
    {
        int flags, n, error;
        socklen_t len;
        fd_set rset, wset;
        struct timeval tval;
     
        flags = Fcntl(sockfd, F_GETFL, 0);
        Fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
     
        error = 0;
        if ( (n = connect(sockfd, saptr, salen)) < 0)
            if (errno != EINPROGRESS)//此错误为期望的错误,表示连接建立已经启动但尚未成功
                return(-1);
     
        /* Do whatever we want while the connect is taking place. */
     
        if (n == 0)
            goto done;    /* connect completed immediately */
     
        FD_ZERO(&rset);
        FD_SET(sockfd, &rset);
        wset = rset;
        tval.tv_sec = nsec;
        tval.tv_usec = 0;
     
        if ( (n = Select(sockfd+1, &rset, &wset, NULL,
                         nsec ? &tval : NULL)) == 0) {
            close(sockfd);        /* timeout */
            errno = ETIMEDOUT;
            return(-1);
        }
     
        if (FD_ISSET(sockfd, &rset) || FD_ISSET(sockfd, &wset)) {
            len = sizeof(error);
            if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &len) < 0)
                return(-1);            /* Solaris pending error */
        } else
            err_quit("select error: sockfd not set");
     
    done:
        Fcntl(sockfd, F_SETFL, flags);    /* restore file status flags */
     
        if (error) {
            close(sockfd);        /* just in case */
            errno = error;
            return(-1);
        }
        return(0);
    }

      如何判断非阻塞connect调用后,连接是否成功建立:

    1. 调用getpeername代替getsockopt,先判断套接字可读或可写条件,如果getpeername以ENOTCONN错误失败返回,那么连接已经建立失败,接着必须以SO_ERROR调用getsockopt,检查套接字上是否存在待处理错误。
    2. 以值为0的长度参数调用read,如果read失败,则connect失败,read返回错误errno给出原因,如果建立连接成功,read返回0.
    3. 再调用connect一次,他应该失败,如果错误是EISCONN,那么套接字已经成功连接,也就是第一次成功连接。

    被中断的connect

      如果对于正常阻塞的套接字,其上的connect在调用TCP三路握手完成前被中断(比如捕捉了某个信号),则connect中断不能由内核自动重启,将返回EINTR,我们不能再次调用connect,这样做导致EADDRINUSE错误。

      这种情况下我们只能再次调用select,相对于非阻塞connect那样,连接建立完成时,select返回可读条件,建立连接失败时select返回套接字即可读又可写。

    非阻塞accept

    客户端

        /*
            初始化套接字操作
        */
    
        connect(sockfd,(sockaddr*)&servAddr,sizeof(servAddr));
        
        struct linger ling;
        ling.l_onoff=1;//cause RST to sent on close()
        ling.l_linger=0;
        setsockopt(sockfd,SOL_SOCKET,SO_LINGER,&ling,sizeof(ling));
        close(sockfd);

    服务器端

    1. if(FD_ISSET(listenfd,&rest))//new connection { sleep(5); clien=sizeof(cliAddr); connfd=accept(listenfd,(sockaddr*)&cliAddr,&clien); }
    1. select向服务器返回可读,但是过一段时间在调用accept
    2. 在服务器从select返回到调用accept期间,服务器TCP收到来自客户端的RST
    3. 这个已完成得连接被服务器TCP驱除队列,假设队列中没有其他的已完成的连接
    4. 服务器端调用accept,但是没有已完成的连接,服务器会一直阻塞在accept上,直到其他某个客户建立连接

    解决办法:

    1. 当使用select获悉某个监听套接字上何时有已完成连接准备被accept时,总是把这个监听套接字设置为非阻塞
    2. 在后续的accept调用中忽略以下错误:EWOULDBLOCK(源自bk,客户终止连接时),ECONNABORTED(POSIX实现,客户终止连接时,EPROTO(SVR4实现,客户中止连接时)和EINTR(如果有信号被捕获)
  • 相关阅读:
    将图片制作数据拒
    python中PIL.Image,OpenCV,Numpy图像格式相互转换
    9、【Hive】初始化元数据库时失败,遇到org.apache.hadoop.hive.metastore.HiveMetaException: Schema initialization FAILED! Metastore state would be inconsistent
    8、【Hive Mysql】MySQL8 提示Public Key Retrieval is not allowed错误解决方法
    7.【Hive】hive启动出现权限错误 /tmp/hive on HDFS should be writable
    一、Zookeeper简明笔记
    6、System times on machines may be out of sync. Check system time and time zones
    5、利用搭建的hadoop集群计算wordcount时出现内存问题
    4、利用hadoop自带的mr 示例测试运行任务失败: Could not find or load main class org.apache.hadoop.mapreduce.v2.app.MRAppMaster
    3、hadoop运行jar包报错 : "/bin/bash: /bin/java: No such file or directory"
  • 原文地址:https://www.cnblogs.com/tianzeng/p/11930057.html
Copyright © 2011-2022 走看看