流？I/O操做？阻塞？epoll?

時間 2020-03-13

標籤阻塞 epoll 简体版

原文原文鏈接

1、流？I/O操做? 阻塞？

(1) 流

能夠進行I/O操做的內核對象
文件、管道、套接字……
流的入口：文件描述符(fd)

(2) I/O操做

全部對流的讀寫操做，咱們均可以稱之爲IO操做。git

當一個流中，在沒有數據read的時候，或者說在流中已經寫滿了數據，再write，咱們的IO操做就會出現一種現象，就是阻塞現象，以下圖。github

(3) 阻塞

阻塞場景: 你有一份快遞，家裏有個座機，快遞到了主動給你打電話，期間你能夠休息。golang

非阻塞，忙輪詢場景: 你性子比較急躁，每分鐘就要打電話詢問快遞小哥一次，到底有沒有到，快遞員接你電話要中止運輸，這樣很耽誤快遞小哥的運輸速度。編程

阻塞等待

空出大腦能夠安心睡覺, 不影響快遞員工做（不佔用CPU寶貴的時間片）。安全

非阻塞，忙輪詢

浪費時間，浪費電話費，佔用快遞員時間（佔用CPU，系統資源）。服務器

很明顯，阻塞等待這種方式，對於通訊上是有明顯優點的，那麼它有哪些弊端呢？微信

2、解決阻塞死等待的辦法

阻塞死等待的缺點

也就是同一時刻，你只能被動的處理一個快遞員的簽收業務，其餘快遞員打電話打不進來，只能乾瞪眼等待。那麼解決這個問題，家裏多買N個座機，可是依然是你一我的接，也處理不過來，須要用影分身術建立都個本身來接電話(採用多線程或者多進程）來處理。多線程

這種方式就是沒有多路IO複用的狀況的解決方案，可是在單線程計算機時代(沒法影分身)，這簡直是災難。併發

那麼若是咱們不借助影分身的方式(多線程/多進程)，該如何解決阻塞死等待的方法呢？負載均衡

辦法一：非阻塞、忙輪詢

while true {
    for i in 流[] {
        if i has 數據 {
            讀 或者 其餘處理
        }
    }
}

非阻塞忙輪詢的方式，可讓用戶分別與每一個快遞員取得聯繫，宏觀上來看，是同時能夠與多個快遞員溝通(併發效果)、可是快遞員在於用戶溝通時耽誤前進的速度(浪費CPU)。

辦法二：select

咱們能夠開設一個代收網點，讓快遞員所有送到代收點。這個網店管理員叫select。這樣咱們就能夠在家休息了，麻煩的事交給select就行了。當有快遞的時候，select負責給咱們打電話，期間在家休息睡覺就行了。

但select 代收員比較懶，她記不住快遞員的單號，還有快遞貨物的數量。她只會告訴你快遞到了，可是是誰到的，你須要挨個快遞員問一遍。

while true {
    select(流[]); //阻塞

  //有消息抵達
    for i in 流[] {
        if i has 數據 {
            讀 或者 其餘處理
        }
    }
}

辦法三：epoll

epoll的服務態度要比select好不少，在通知咱們的時候，不只告訴咱們有幾個快遞到了，還分別告訴咱們是誰誰誰。咱們只須要按照epoll給的答覆，來詢問快遞員取快遞便可。

while true {
    可處理的流[] = epoll_wait(epoll_fd); //阻塞

  //有消息抵達，所有放在 「可處理的流[]」中
    for i in 可處理的流[] {
        讀 或者 其餘處理
    }
}

3、epoll？

與select，poll同樣，對I/O多路複用的技術
只關心「活躍」的連接，無需遍歷所有描述符集合
可以處理大量的連接請求(系統能夠打開的文件數目)

4、epoll的API

(1) 建立EPOLL

/** 
 * @param size 告訴內核監聽的數目 
 * 
 * @returns 返回一個epoll句柄（即一個文件描述符） 
 */
int epoll_create(int size);

使用

int epfd = epoll_create(1000);

建立一個epoll句柄，其實是在內核空間，創建一個root根節點，這個根節點的關係與epfd相對應。

(2) 控制EPOLL

/**
* @param epfd 用epoll_create所建立的epoll句柄
* @param op 表示對epoll監控描述符控制的動做
*
* EPOLL_CTL_ADD(註冊新的fd到epfd)
* EPOLL_CTL_MOD(修改已經註冊的fd的監聽事件)
* EPOLL_CTL_DEL(epfd刪除一個fd)
*
* @param fd 須要監聽的文件描述符
* @param event 告訴內核須要監聽的事件
*
* @returns 成功返回0，失敗返回-1, errno查看錯誤信息
*/
int epoll_ctl(int epfd, int op, int fd,
struct epoll_event *event);


struct epoll_event {
    __uint32_t events; /* epoll 事件 */
    epoll_data_t data; /* 用戶傳遞的數據 */
}

/*
 * events : {EPOLLIN, EPOLLOUT, EPOLLPRI,
                         EPOLLHUP, EPOLLET, EPOLLONESHOT}
 */
typedef union epoll_data {
    void *ptr;
    int fd;
    uint32_t u32;
    uint64_t u64;
} epoll_data_t;

使用

struct epoll_event new_event;

new_event.events = EPOLLIN | EPOLLOUT;
new_event.data.fd = 5;

epoll_ctl(epfd, EPOLL_CTL_ADD, 5, &new_event);

建立一個用戶態的事件，綁定到某個fd上，而後添加到內核中的epoll紅黑樹中。

(3) 等待EPOLL

/**
*
* @param epfd 用epoll_create所建立的epoll句柄
* @param event 從內核獲得的事件集合
* @param maxevents 告知內核這個events有多大,
* 注意: 值 不能大於建立epoll_create()時的size.
* @param timeout 超時時間
* -1: 永久阻塞
* 0: 當即返回，非阻塞
* >0: 指定微秒
*
* @returns 成功: 有多少文件描述符就緒,時間到時返回0
* 失敗: -1, errno 查看錯誤
*/
int epoll_wait(int epfd, struct epoll_event *event,
                             int maxevents, int timeout);

使用

struct epoll_event my_event[1000];

int event_cnt = epoll_wait(epfd, my_event, 1000, -1);

epoll_wait是一個阻塞的狀態，若是內核檢測到IO的讀寫響應，會拋給上層的epoll_wait, 返回給用戶態一個已經觸發的事件隊列，同時阻塞返回。開發者能夠從隊列中取出事件來處理，其中事件裏就有綁定的對應fd是哪一個(以前添加epoll事件的時候已經綁定)。

(4) 使用epoll編程主流程骨架

int epfd = epoll_crete(1000);

//將 listen_fd 添加進 epoll 中
epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd,&listen_event);

while (1) {
    //阻塞等待 epoll 中 的fd 觸發
    int active_cnt = epoll_wait(epfd, events, 1000, -1);

    for (i = 0 ; i < active_cnt; i++) {
        if (evnets[i].data.fd == listen_fd) {
            //accept. 而且將新accept 的fd 加進epoll中.
        }
        else if (events[i].events & EPOLLIN) {
            //對此fd 進行讀操做
        }
        else if (events[i].events & EPOLLOUT) {
            //對此fd 進行寫操做
        }
    }
}

九、Golang中的Channel底層深度剖析

首先聲明，本文不介紹channel的基礎語法和使用場景，若是想golang中的channel的基礎語法其餘地方有不少地方介紹。這裏只介紹channel的一些底層實現原理的剖析。

(1)Channel特性

首先，咱們先複習一下Channel都有哪些特性？

給一個 nil channel 發送數據，形成永遠阻塞
從一個 nil channel 接收數據，形成永遠阻塞
給一個已經關閉的 channel 發送數據，引發 panic
從一個已經關閉的 channel 接收數據，若是緩衝區中爲空，則返回一個零值
無緩衝的channel是同步的，而有緩衝的channel是非同步的

以上5個特性是死東西，也能夠經過口訣來記憶：「空讀寫阻塞，寫關閉異常，讀關閉空零」。

下面以簡單的示例來演示Go如何經過channel來實現通訊。

package main
import (
    "fmt"
    "time"
)
func goRoutineA(a <-chan int) {
    val := <-a
    fmt.Println("goRoutineA received the data", val)
}
func goRoutineB(b chan int) {
    val := <-b
    fmt.Println("goRoutineB  received the data", val)
}
func main() {
    ch := make(chan int, 3)
    go goRoutineA(ch)
    go goRoutineB(ch)
    ch <- 3
    time.Sleep(time.Second * 1)
}

5、epoll的觸發模式

(1) 水平觸發

水平觸發的主要特色是，若是用戶在監聽epoll事件，當內核有事件的時候，會拷貝給用戶態事件，可是若是用戶只處理了一次，那麼剩下沒有處理的會在下一次epoll_wait再次返回該事件。

這樣若是用戶永遠不處理這個事件，就致使每次都會有該事件從內核到用戶的拷貝，耗費性能，可是水平觸發相對安全，最起碼事件不會丟掉，除非用戶處理完畢。

(2) 邊緣觸發

邊緣觸發，相對跟水平觸發相反，當內核有事件到達，只會通知用戶一次，至於用戶處理仍是不處理，之後將不會再通知。這樣減小了拷貝過程，增長了性能，可是相對來講，若是用戶馬虎忘記處理，將會產生事件丟的狀況。

5、簡單的epoll服務器(C語言)

(1) 服務端

#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)
{
    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);

    // 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));

    // listen
    listen(listen_fd, 10);

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

    // listen_fd -> epoll
    event.events = EPOLLIN;
    event.data.fd = listen_fd;
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event) < 0) {
        perror("epoll ctl add listen_fd ");
        goto END;
    }

    my_events = malloc(sizeof(struct epoll_event) * EPOLL_MAX_NUM);


    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;
}

(2) 客戶端

#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 (7778)

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);
    }


    // 建立socket
    if ( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        fprintf(stderr, "socket error");
        exit(0);
    }


    // server addr 賦值
    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);
    }


    // 連接服務端
    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;



    // 不斷的從標準輸入字符串
    while (fgets(input, 100, stdin) != NULL)
    {
        printf("[send] %s\n", input);
        n = 0;
        // 把輸入的字符串發送 到 服務器中去
        n = send(sockfd, input, strlen(input), 0);
        if (n < 0) {
            perror("send");
        }

        n = 0;
        count = 0;


        // 讀取 服務器返回的數據
        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;
}