1
2
3
4
5
6
7

list_for_each_entry_safe(curr, next, &q->task_list, task_list) {
    unsigned flags = curr->flags;
    if (curr->func(curr, mode, wake_flags, key) &&
           (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
         break;
   }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

#include <arpa/inet.h>
#include <assert.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>

#define SERVER_ADDRESS "0.0.0.0"
#define SERVER_PORT 10086
#define WORKER_COUNT 4
char sendbuff[1024];

int worker_process(int listenfd, int i) {
    while (1) {
        printf("I am work %d, my pid is %d, begin to accept connections \n", i, getpid());
        struct sockaddr_in client_info;
        socklen_t client_info_len = sizeof(client_info);
        int connfd = accept(listenfd, (struct sockaddr *)&client_info, &client_info_len);
        if (connfd != -1) {
            printf("worker %d accept success\n", i);
            printf("ip :%s\t", inet_ntoa(client_info.sin_addr));
            printf("port: %d \n", client_info.sin_port);
            snprintf(sendbuff, sizeof(sendbuff), "accept PID is %d\n", getpid());
            send(connfd, sendbuff, strlen(sendbuff) + 1, 0);
            // Process data or perform tasks with the connection
            // ...
            close(connfd);
            printf("worker %d close success\n", i);
        }
        else {
            printf("worker %d accept failed", i);
        }
    }
}
int main() {
    int i;
    struct sockaddr_in address;
    bzero(&address, sizeof(address));
    address.sin_family = AF_INET;
    inet_pton(AF_INET, SERVER_ADDRESS, &address.sin_addr);
    address.sin_port = htons(SERVER_PORT);
    int listenfd = socket(PF_INET, SOCK_STREAM, 0);
    // Add the SO_REUSEADDR option
    int optval = 1;
    setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
    setsockopt(listenfd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval));

    int ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address));
    printf("bind ret=%d\n", ret);
    ret = listen(listenfd, 5);
    for (i = 0; i < WORKER_COUNT; i++) {
        printf("Create worker %d\n", i + 1);
        pid_t pid = fork();
        /*child  process */
        if (pid == 0) {
            worker_process(listenfd, i);
        }
        if (pid < 0) {
            printf("fork error");
        }
    }

    //wait child process
    int status;
    wait(&status);
    return 0;

}

1

strace -f

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <netdb.h>
#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdlib.h>
#include <sys/wait.h>

#define PROCESS_NUM 10
#define MAXEVENTS 64

static int create_and_bind(char* port) {
    int fd = socket(PF_INET, SOCK_STREAM, 0);
    struct sockaddr_in serveraddr;
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
    serveraddr.sin_port = htons(atoi(port));
    bind(fd, (struct sockaddr *)&serveraddr, sizeof(serveraddr));
    return fd;
}

static int make_socket_non_blocking(int sfd) {
    int flags, s;

    flags = fcntl(sfd, F_GETFL, 0);
    if (flags == -1) {
        perror("fcntl");
        return -1;
    }

    flags |= O_NONBLOCK;
    s = fcntl(sfd, F_SETFL, flags);
    if (s == -1) {
        perror("fcntl");
        return -1;
    }

    return 0;
}

int main() {
    int sfd, s;
    int efd;
    struct epoll_event event;
    struct epoll_event* events;

    sfd = create_and_bind("8888");
    if (sfd == -1) {
        abort();
    }

    s = make_socket_non_blocking(sfd);
    if (s == -1) {
        abort();
    }

    s = listen(sfd, SOMAXCONN);
    if (s == -1) {
        perror("listen");
        abort();
    }

    efd = epoll_create(MAXEVENTS);
    if (efd == -1) {
        perror("epoll_create");
        abort();
    }

    event.data.fd = sfd;
    // event.events = EPOLLIN | EPOLLET;
    event.events = EPOLLIN;
    s = epoll_ctl(efd, EPOLL_CTL_ADD, sfd, &event);
    if (s == -1) {
        perror("epoll_ctl");
        abort();
    }

    /* Buffer where events are returned */
    events = calloc(MAXEVENTS, sizeof event);
    int k;
    for (k = 0; k < PROCESS_NUM; k++) {
        int pid = fork();
        if (pid == 0) {
            /* The event loop */
            while (1) {
                int n, i;
                n = epoll_wait(efd, events, MAXEVENTS, -1);
                printf("process %d return from epoll_wait!\n", getpid());
                /* sleep here is very important!*/
                // sleep(2);
                for (i = 0; i < n; i++) {
                    if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP) || (!(events[i].events & EPOLLIN))) {
                        /* An error has occured on this fd, or the socket is not
                        ready for reading (why were we notified then?) */
                        fprintf(stderr, "epoll error\n");
                        close(events[i].data.fd);
                    }
                    else if (sfd == events[i].data.fd) {
                        /* We have a notification on the listening socket, which
                        means one or more incoming connections. */
                        struct sockaddr in_addr;
                        socklen_t in_len;
                        int infd;
                        char hbuf[NI_MAXHOST], sbuf[NI_MAXSERV];

                        in_len = sizeof in_addr;
                        infd = accept(sfd, &in_addr, &in_len);
                        if (infd == -1) {
                            printf("process %d accept failed!\n", getpid());
                            break;
                        }
                        printf("process %d accept successed!\n", getpid());

                        /* Make the incoming socket non-blocking and add it to the
                        list of fds to monitor. */
                        close(infd);
                    }
                }
            }
        }
    }
    int status;
    wait(&status);
    free(events);
    close(sfd);
    return EXIT_SUCCESS;
}

1

telnet localhost 8888

1
2
3
4
5
6

process 545913 return from epoll_wait!
process 545912 return from epoll_wait!
process 545913 accept successed!
process 545912 accept failed!
process 545911 return from epoll_wait!
process 545911 accept failed!

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// fs/eventpoll.c 精简
struct eventpoll {
    // 阻塞在epoll_wait的task的睡眠队列
    wait_queue_head_t wq;
    // 存在就绪文件句柄的list，该list上的文件句柄事件将会全部上报给应用
    struct list_head rdllist;
    // 存放加入到此epoll句柄的文件句柄的红黑树容器
    struct rb_root rbr;
    // 该epoll结构对应的文件句柄，应用通过它来操作该epoll结构
    struct file *file;
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

struct epitem {
    // 该字段链接入epoll句柄的红黑树容器
    struct rb_node rbn;
    // 当该文件句柄有事件发生时，该字段链接入“就绪链表”，准备上报给用户态
    struct list_head rdllink;
    // 封装实际的文件
    struct epoll_filefd {
        struct file *file;
        int fd;
    } ffd;
    // 反向指向其所属的epoll句柄
    struct eventpoll *ep;
};

1
2
3
4
5
6
7

struct __wait_queue {
    unsigned int flags;
    // 至于这个private到底是什么，内核并不限制，显然，它可以是task，也可以是别的。
    void *private;
    wait_queue_func_t func;
    struct list_head task_list;
};

1
2
3
4

struct __wait_queue_head {
    spinlock_t lock;
    struct list_head task_list;
};

1
2
3

static inline wait_queue_head_t *sk_sleep(struct sock *sk){
    return &rcu_dereference_raw(sk->sk_wq)->wait;
}