Understand IO reuse - blocking and non-blocking

Understand IO reuse – blocking and non-blocking

Flow:

A kernel object that can perform I/O operations
This includes files, pipes, sockets, and so on
The entrance of a stream:File descriptor(fd)

Streams are oriented, read and write

If you don’t have data yet, if you try to read it, it will block here.

Let’s say you need soap to wash your socks, and there’s no soap in the house. There’s a delivery guy who delivers soap to you, and there’s nothing to do until it arrives, which is blocked.

We can also try the non-blocking approach of polling all the time.

Blocking vs non-blocking:

Block wait: the time slice that does not occupy the CPU. When the process or thread is ready and the resource is no longer blocked, the process or thread scheduling can let it continue to perform \\ non-blocking polling: occupying the CPU, it will continue to poll the resource if it is ready, and then execute the following program

In most cases, blocking wait is a better choice because it does not consume CPU resources and the resource will continue to execute when it is ready.

Disadvantages of blocking wait:

Blocking wait is generally the best option if you are waiting for a resource. But if you have to wait for more than one resource at a time, the blocking method will have to wait one by one, which is actually less efficient. In this case, non-blocking polling, that is, polling all resources each time, if there is one ready, go to the processing

In general scenarios, blocking wait will be selected, but because it cannot handle multiple concurrent requests well, the same blocking can only handle one stream of blocking listening at a time, at this point, we will choose multiplexing IO:

Can block wait without wasting CPU resources

Monitor the status of multiple I/O requests at the same time

Problems solved by IO multiplexing

How to solve a large number of I/O read and write requests without wasting CPU?

`Method 1: blocking + multithreading/multi-process`

Let each thread handle an I/O request without wasting CPU resources because each thread is blocked waiting, but the disadvantages are obvious:

Multithreading, or multithreading, consumes resources, and thread switching is expensive

`Method 2: Non-blocking + Busy polling`

Method 2 can solve a large number of I/O requests, but consumes CPU resources

// Implementation of non-blocking busy polling
while true{
    forI flow in [] {// Walk through all streams at once
        ifI has data {read or other processing}}}Copy the code

`Method 3: select`

Our select will manage all the I/O requests on our behalf. If an I/O resource has arrived, select will notify us of the current I/O resource, but only if the resource has arrived.

Select has limited resources. A maximum of 1024 I/O status can be monitored

In short: a call to SELECT blocks there and does not return until the resource is ready

while true{the select (flow []);/ / blocking
    // If a resource arrives, it returns from blocking and polls all streams
	forI flow in [] {ifI has data {read or other processing}}}Copy the code

`Method 4: epoll`

Epoll is a much more powerful tool than Select in that it not only tells us which packages arrived, but also which ones arrived, so we don’t have to go through all the streams to see which ones are ready. Also, epoll can handle much larger IO requests than SELECT

while true{processable flow [] = epoll_wait(epoll_fd);/ / blocking
    
    forI in processable streams []{read or other processing}}Copy the code

Epoll API and internal mechanism

1. Create epoll (epoll_create)

int epoll_create(int size);	//size: the number of kernel listeners
// Return an epoll handle
Copy the code

All epoll_CREATE (size) does is create a root node of a red-black tree in the kernel, which is returned to the user

2. Control epoll (epoll_ctl)

Param op op is an optional parameter. * EPOLL_CTL_ADD registers a new FD into an EPFD. * EPOLL_CTL_MOD modifs the listener event of a registered FD EPOLL_CTL_DEL indicates that epfd deletes a fd * @3.param fd fd is the file descriptor to listen for * @4.param event tells the kernel the event to listen for * * @returns 0 on success, -1 on failure,errno Views error information * /
int epoll_ctl(int epfd,int op,int fd,strucct epoll_event *event);

Copy the code

struct epoll_event:

struct epoll_event{
    __uint32_t events;	/ / epoll event
    epoll_data_t data;	// Data passed by the user
}
typedef union epoll_data{
    void *ptr;
    int fd;
    uint32_t u32;
    uint64_t u64;
}
Copy the code

Example:

struct epoll_event new_event;
new_event.events = EPOLLIN | EPOLLOUT;	// Readable and writable events
new_evnet.data.fd = 5;	//event binds the file descriptor 5

epoll_ctl(epfd,EPOLL_CLT_ADD,5,&new_event);// Add file descriptor 5 as a read-write event to the root node
Copy the code

With epoll\_ctl add an event reading/writing file descriptor 5 to our epoll, \\ will insert a new\_event into the tree of the Kernel. \\ can then walk through the tree from the Kernel while epoll\_wait to find all events that trigger these events. And return it to the user.

3. Wait for the epoll (epoll_wait)

/** ** @param epfd Epoll handle created with epoll_create * @param event A collection of events (not passed in data, but used to get triggered events) * @param maxEvents Tell the kernel how big events are (not > size at creation time) * @param Timeout Timeout time * -1: permanently blocked * 0: return immediately, non-blocking * >0: specify microseconds * * @returns successful: return number of file descriptors ready, and return 0 when the time is up *. Failure :-1,errno view error */

int epoll_wait(int epfd,struct epoll_event *event,int maxevents,int timeout);
Copy the code

Example:

struct epoll_event my_event[1000].int event_cnt = epoll_wait(epfd,my_event,1000.- 1);
Copy the code

If epoll_wait returns, all events will be placed in my_event, and event_cnt returns the number of events, so we can use the my_event array and the number of events told by event_cnt to iterate over my_event [0, event_cnT-1]. Deal with this.

Epoll programming architecture

Epollint epfd = epfd_create(1000); Epoll_ctl (epfd,EPOLL_CTL_ADD,listen_fd,&listen_event); Int active_cnt = epoll_wait(epfd,events,1000,-1); For (int I = 0; i < active_cnt; If (events[I].data.fd == list.fd){if (events[I].data.fd == list.fd){if (events[I].data.fd == list.fd){if (events[I].data.fd == list.fd){if (events[I].data.fd == list.fd){ Add the accept fd to epoll} else if (events[I].events & EPOLLIN) else if (events[I].events & EPOLLIN (events[I].events & EPOLLOUT)
Copy the code

The following is a common framework of epoll seen in a blog:

 1 for(; ;)2     { 3         nfds = epoll_wait(epfd,events,20.500); 4         for(i=0; i<nfds; ++i)5         { 6             if(events[i].data.fd==listenfd) 7 {8 connfd = accept(listenfd,(sockaddr *)& clientAddr, &clilen); 9 ev.data.fd=connfd; 10 ev.events=EPOLLIN|EPOLLET; 11 epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev); Elseif (events[I].events&epollin) else if(events[I].events&epollin) Socket14 {15 n = read(sockfd, line, MAXLINE)) < 0 // read 16 ev.data.ptr = md; / / md for custom type, add data 17 ev. Events = EPOLLOUT | EPOLLET; 18 epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); 19}20 else if(events[I].events&epollout) Socket21 {22 struct myepoll_data* md = (myepoll_data*)events[I].data.ptr; Sockfd = md->fd; 24 send( sockfd, md->ptr, strlen((char*)md->ptr), 0 ); 25 ev.data.fd=sockfd; 26 ev.events=EPOLLIN|EPOLLET; 27 epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev); Else30 {31} 32}33}34}
Copy the code

Here is another example of code used by epoll in the blog epoll TCP server and client – Jianshu.com

socket_server.cpp:

#include <fcntl.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/epoll.h>

using namespace std;

const int MAX_EPOLL_EVENTS = 1000;
const int MAX_MSG_LEN = 1024;

void setFdNonblock(int fd)
{
    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);
}

void err_exit(const char *s){
    printf("error: %s\n",s);
    exit(0);
}

int create_socket(const char *ip, const int port_number)// Create a listening port
{
    struct sockaddr_in server_addr = {0};
    /* Set the ipv4 mode */
    server_addr.sin_family = AF_INET;           /* ipv4 */
    /* Set the port number */
    server_addr.sin_port = htons(port_number);
    /* Set the host address */
    if(inet_pton(server_addr.sin_family, ip, &server_addr.sin_addr) == - 1) {err_exit("inet_pton");
    }
    /* Set up socket */
    int sockfd = socket(PF_INET, SOCK_STREAM, 0);
    if(sockfd == - 1) {err_exit("socket");
    }
    /* Set the reuse mode */
    int reuse = 1;
    if(setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) == - 1)
    {
        err_exit("setsockopt");
    }
    /* Bond port */
    if(bind(sockfd, (sockaddr *)&server_addr, sizeof(server_addr)) == - 1) {err_exit("bind");
    }
    /* Set passive open */
    if(listen(sockfd, 5) = =- 1) {err_exit("listen");
    }
    return sockfd;
}

int main(int argc, const char *argv[])
{
    / * help * /
    if(argc < 3) {printf("usage:%s ip port\n", argv[0]);
        exit(0);
    }
    /* Get server parameters */
    const char * ip = argv[1];
    const int port = atoi(argv[2]);
    /* Create a socket */
    int sockfd = create_socket(ip, port);
    printf("success create sockfd %d\n", sockfd);
    setFdNonblock(sockfd);
    /* Create epoll */
    int epollfd = epoll_create1(0);
    if(epollfd == - 1) err_exit("epoll_create1");
    /* Add sockfd to ePollfd interest list */
    struct epoll_event ev;
    ev.data.fd = sockfd;
    ev.events = EPOLLIN ;
    if(epoll_ctl(epollfd, EPOLL_CTL_ADD, sockfd, &ev) == - 1) {err_exit("epoll_ctl1");
    }
    /* Create a list of events returned by wait */
    struct epoll_event events[MAX_EPOLL_EVENTS] = {0};
    /* Start waiting for all events mounted on epoll */

    while(1) {/* Wait for events */
        printf("begin wait\n");
        int number = epoll_wait(epollfd, events, MAX_EPOLL_EVENTS, - 1);
        printf("end wait\n");
        sleep(1);
        if(number > 0) {/* Iterate over all events */
            for (int i = 0; i < number; i++)
            {
                int eventfd = events[i].data.fd;
                /* If the fd that triggers the event is sockfd, then someone is connected and we need to accept it */
                if(eventfd == sockfd){
                    printf("accept new client... \n");
                    struct sockaddr_in client_addr;
                    socklen_t client_addr_len = sizeof(client_addr);
                    int connfd = accept(sockfd, (struct sockaddr *)&client_addr, &client_addr_len);
                    setFdNonblock(connfd);
                    After /* accept, the file descriptor needs to be added to the listener list */
                    struct epoll_event ev;
                    ev.data.fd = connfd;
                    ev.events = EPOLLIN;
                    if(epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &ev) == - 1) {err_exit("epoll_ctl2");
                    }
                    printf("accept new client end.\n");
                }
                /* If the triggered fd is not a sockfd, it is the new connfd */
                else{
                    /* Read out the contents until a carriage return is reached. The content is then displayed. * /
                    printf("read start... \n");
                    while(1) {char buff = - 1;
                        int ret = read(eventfd, &buff, 1);
                        if(ret > 0) {printf("%c", buff);
                        }
                        if(buff == '\n') {break;
                        }
                        else if (ret == 0) {printf("client close.\n");
                            close(eventfd);
                            epoll_ctl(epollfd, EPOLL_CTL_DEL, eventfd, NULL);
                            break;
                        }
                        else if (ret < 0) {printf("read error.\n");
                            break; }}printf("read end.\n");
                }
            }
        }
    }
}

Copy the code

As you can see, the server does the following:

1. Preparation Stage:

int sockfd = create_socket(ip, port);	// Create the file descriptor sockfd for listening connections
Copy the code

int epollfd = epoll_create1(0);			/ / create epoll
Copy the code

// Add the IN event of the listening file descriptor to epoll
struct epoll_event ev;
ev.data.fd = sockfd;
ev.events = EPOLLIN ;
if(epoll_ctl(epollfd, EPOLL_CTL_ADD, sockfd, &ev) == - 1) {err_exit("epoll_ctl1");
}
Copy the code

    /* Create an array to hold events returned by wait */
    struct epoll_event events[MAX_EPOLL_EVENTS] = {0};
Copy the code

2. Main loop

    while(1) {/* Wait for events */
        printf("begin wait\n");
        int number = epoll_wait(epollfd, events, MAX_EPOLL_EVENTS, - 1);
        printf("end wait\n");
        sleep(1);
        if(number > 0) {/* Iterate over all events */
            for (int i = 0; i < number; i++)
            {
                int eventfd = events[i].data.fd;
                /* If the fd that triggers the event is sockfd, then someone is connected and we need to accept it */
                if(eventfd == sockfd){
                    printf("accept new client... \n");
                    struct sockaddr_in client_addr;
                    socklen_t client_addr_len = sizeof(client_addr);
                    int connfd = accept(sockfd, (struct sockaddr *)&client_addr, &client_addr_len);
                    setFdNonblock(connfd);
                    After /* accept, the file descriptor needs to be added to the listener list */
                    struct epoll_event ev;
                    ev.data.fd = connfd;
                    ev.events = EPOLLIN;
                    if(epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &ev) == - 1) {err_exit("epoll_ctl2");
                    }
                    printf("accept new client end.\n");
                }
                /* If the triggered fd is not a sockfd, it is the new connfd */
                else{
                    /* Read out the contents until a carriage return is reached. The content is then displayed. * /
                    printf("read start... \n");
                    while(1) {char buff = - 1;
                        int ret = read(eventfd, &buff, 1);
                        if(ret > 0) {printf("%c", buff);
                        }
                        if(buff == '\n') {break;
                        }
                        else if (ret == 0) {printf("client close.\n");
                            close(eventfd);
                            epoll_ctl(epollfd, EPOLL_CTL_DEL, eventfd, NULL);
                            break;
                        }
                        else if (ret < 0) {printf("read error.\n");
                            break; }}printf("read end.\n"); }}}Copy the code

socket_client.cpp:

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <string>
#include <iostream>

using namespace std;

void err_exit(const char *s){
    printf("error: %s\n",s);
    exit(0);
}

int create_socket(const char *ip, const int port_number)
{
    struct sockaddr_in server_addr = {0};
    /* Set the ipv4 mode */
    server_addr.sin_family = AF_INET;           /* ipv4 */
    /* Set the port number */
    server_addr.sin_port = htons(port_number);
    /* Set the host address */
    if(inet_pton(PF_INET, ip, &server_addr.sin_addr) == - 1) {err_exit("inet_pton");
    }

    /* Set up socket */
    int sockfd = socket(PF_INET, SOCK_STREAM, 0);
    if(sockfd == - 1) {err_exit("socket");
    }

    if(connect(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr)) == - 1) {err_exit("connect");
    }

    return sockfd;
}

int main(int argc, const char *argv[]){
    if(argc < 3) {printf("usage:%s ip port\n", argv[0]);
        exit(0);
    }
    /* Get server parameters */
    const char * ip = argv[1];
    const int port = atoi(argv[2]);
    // Create socket
    int sock = create_socket(ip, port);
    // Make a request to the server (specific IP and port)
    
    while(1){
        string buff;
        getline(cin, buff);
        if(buff == "exit") break;
        write(sock, buff.c_str(), buff.size());
        char end = '\n';
        write(sock, &end, 1);
    }
    close(sock);
    return 0;
}
Copy the code

g++ -Wall socket_server.cpp -o server && g++ -Wall socket_client.cpp -o client
Copy the code

./server localhost 1234
Copy the code

./client localhost 1234
Copy the code

If you type text on the client and press Enter, it will appear on the server. Press Ctrl + C to close the client or type Exit to close the client.

The server first creates a passive open socket file descriptor and then adds the file descriptor to the epoll interest list. Then we go into the loop. Whenever interest list wait ends, the corresponding file descriptor can be operated on. If a client is connected to the passively opened socket file descriptor, the passively opened socket file descriptor can be accepted. The new file descriptor created after accept is the file descriptor that communicates with the client and continues to be added to the list of interests. When the client sends data, the file descriptor also generates a readable signal, which causes the wait to end and the data sent by the client to be read and displayed in processing mode.

Epoll level trigger with edge trigger

Level trigger

Horizontal trigger:

When the first epoll_wait is performed, both A and B events are triggered. Let’s say that when we process event A, we read half of it, we haven’t finished processing it, and the other half is left in event_A. So if it’s horizontal, event_A will be returned next time until it’s actually processed or deleted.

So this ensures the integrity and security of handling the event.

Disadvantages: If the user does not process it, event_A is returned every time, which involves syscall system calls and can be performance costly.

Edge trigger

Edge trigger:

In the first epoll_wait, both A and B events are triggered, so we get A and B events. Edge triggering is whether you actually process it or not, even if you don’t read A byte, it will think you are done, so it won’t fire the same A and B event next time.

Advantages: Good performance

Disadvantages: The event may not be handled completely

Their difference is similar to TCP and UDP, TCP sent this data, there is a retransmission mechanism, UDP is sent no matter

Epoll code instance

Here is an example from Liu Danbing

(1) the service side

#include <stdio.h>
#include <stdlib.h>
#include <ctype.h>
#include <string.h>

#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>

#include <sys/epoll.h>

#define SERVER_PORT (7778)
#define EPOLL_MAX_NUM (2048)
#define BUFFER_MAX_LEN (4096)

char buffer[BUFFER_MAX_LEN];

void str_toupper(char *str)	// The entire string is capitalized
{
    int i;
    for (i = 0; i < strlen(str); i ++) {
        str[i] = toupper(str[i]); }}int main(int argc, char **argv)
{
    int listen_fd = 0;
    int client_fd = 0;
    struct sockaddr_in server_addr;
    struct sockaddr_in client_addr;
    socklen_t client_len;

    int epfd = 0;
    struct epoll_event event, *my_events;

    // socket
    listen_fd = socket(AF_INET, SOCK_STREAM, 0);	// Create a socket

    // bind
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(SERVER_PORT);
    bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr));	// Bind listen_fd to the port

    // listen
    listen(listen_fd, 10);	// Set the socket descriptor to a listener descriptor

    // epoll create
    epfd = epoll_create(EPOLL_MAX_NUM);
    if (epfd < 0) {
        perror("epoll create");
        goto END;
    }

    // listen_fd -> epoll
    event.events = EPOLLIN;		/ / read event
    event.data.fd = listen_fd;	// The event file is bound to listen_fd
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event) < 0) {	// Listen_fd adds the read event to epoll
        perror("epoll ctl add listen_fd ");
        goto END;
    }

    my_events = malloc(sizeof(struct epoll_event) * EPOLL_MAX_NUM);	// Create an event[] array


    while (1) {
        // epoll wait
        int active_fds_cnt = epoll_wait(epfd, my_events, EPOLL_MAX_NUM, - 1);	
        int i = 0;
        for (i = 0; i < active_fds_cnt; i++) {
            // if fd == listen_fd
            if (my_events[i].data.fd == listen_fd) {
                //accept
                client_fd = accept(listen_fd, (struct sockaddr*)&client_addr, &client_len);
                if (client_fd < 0) {
                    perror("accept");
                    continue;
                }

                char ip[20];
                printf("new connection[%s:%d]\n", inet_ntop(AF_INET, &client_addr.sin_addr, ip, sizeof(ip)), ntohs(client_addr.sin_port));

                event.events = EPOLLIN | EPOLLET;
                event.data.fd = client_fd;
                epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &event);
            }
            else if (my_events[i].events & EPOLLIN) {
                printf("EPOLLIN\n");
                client_fd = my_events[i].data.fd;

                // do read

                buffer[0] = '\ 0';
                int n = read(client_fd, buffer, 5);
                if (n < 0) {
                    perror("read");
                    continue;
                }
                else if (n == 0) {
                    epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &event);
                    close(client_fd);
                }
                else {
                    printf("[read]: %s\n", buffer);
                    buffer[n] = '\ 0';
#if 1
                    str_toupper(buffer);
                    write(client_fd, buffer, strlen(buffer));
                    printf("[write]: %s\n", buffer);
                    memset(buffer, 0, BUFFER_MAX_LEN);
#endif

                    /* event.events = EPOLLOUT; event.data.fd = client_fd; epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event); * /}}else if (my_events[i].events & EPOLLOUT) {
                printf("EPOLLOUT\n");
                /* client_fd = my_events[i].data.fd; str_toupper(buffer); write(client_fd, buffer, strlen(buffer)); printf("[write]: %s\n", buffer); memset(buffer, 0, BUFFER_MAX_LEN); event.events = EPOLLIN; event.data.fd = client_fd; epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event); * /
            }
        }
    }

END:
    close(epfd);
    close(listen_fd);
    return 0;
}
Copy the code

(2) Client

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>

#define MAX_LINE (1024)
#define SERVER_PORT (7780)

void setnoblocking(int fd)
{
    int opts = 0;
    opts = fcntl(fd, F_GETFL);
    opts = opts | O_NONBLOCK;
    fcntl(fd, F_SETFL);
}

int main(int argc, char **argv)
{
    int sockfd;
    char recvline[MAX_LINE + 1] = {0};

    struct sockaddr_in server_addr;

    if(argc ! =2) {
        fprintf(stderr."usage ./client <SERVER_IP>\n");
        exit(0);
    }


    / / create a socket
    if ( (sockfd = socket(AF_INET, SOCK_STREAM, 0))"0) {
        fprintf(stderr."socket error");
        exit(0);
    }


    // Server addr assigned
    bzero(&server_addr, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(SERVER_PORT);

    if (inet_pton(AF_INET, argv[1], &server_addr.sin_addr) <= 0) {
        fprintf(stderr."inet_pton error for %s", argv[1]);
        exit(0);
    }


    // Connect to the server
    if (connect(sockfd, (struct sockaddr*) &server_addr, sizeof(server_addr)) < 0) {
        perror("connect");
        fprintf(stderr."connect error\n");
        exit(0);
    }

    setnoblocking(sockfd);

    char input[100];
    int n = 0;
    int count = 0;



    // Continuously from the standard input string
    while (fgets(input, 100.stdin) != NULL)
    {
        printf("[send] %s\n", input);
        n = 0;
        // Send the input string to the server
        n = send(sockfd, input, strlen(input), 0);
        if (n < 0) {
            perror("send");
        }

        n = 0;
        count = 0;


        // Read the data returned by the server
        while (1)
        {
            n = read(sockfd, recvline + count, MAX_LINE);
            if (n == MAX_LINE)
            {
                count += n;
                continue;
            }
            else if (n < 0){
                perror("recv");
                break;
            }
            else {
                count += n;
                recvline[count] = '\ 0';
                printf("[recv] %s\n", recvline);
                break; }}}return 0;
}
Copy the code

Results:

Understand IO reuse – blocking and non-blocking