^-^
后端技术_dpdk实现epoll
最近更新：2024-09-23   |   字数总计：1.8k   |   阅读估时：9分钟   |   阅读量：次
  1. EPOLL实现原理
    1. epoll数据结构
      1. why红黑树？
    2. 线程安全机制
      1. 红黑树线程安全
      2. 就绪队列线程安全
    3. epoll如何做到就绪
    4. epoll ET与LT模式
  2. EPOLL实现

EPOLL实现原理

epoll数据结构

why红黑树？
  • 查询频繁型，添加、删除等操作不频繁
  • 强查找型数据结构：hash（占用内存大）、rbtree（综合最优）、链表（越来越慢）、跳表
    1
    2
    3
    int epfd = epoll_create(size); // 创建红黑树
    epoll_ctl(epfd, EPOLL_ADD/DEL/MUD, fd, event); // 操作红黑树，红黑树以fd作为key，所以event可以为空
    epoll_wait(epfd, events, length, timeout); // 就绪队列检索

线程安全机制

红黑树线程安全
  • 锁根节点，因为查找不加锁
  • 添加、删除、改写加mutex锁，一次只允许一个操作
就绪队列线程安全
  • spinlock锁或rcu

epoll如何做到就绪

  • 通过回调机制
    1
    2
    3
    4
    5
    void callbackfunc(){
        1. 通过fd查找节点
        2. 加入就绪队列
        3. 设置就绪标志位
    }
  • 协议栈在特定时候通过回调机制通知epoll：
    1. accept成功时
    2. recvbuf有数据时
    3. sendbuf可写时
    4. fin标志位为1时

epoll ET与LT模式

  • et与lt模式的区别就是回调机制的调用频率问题：如果一次没有处理完recvbuf中的数据，lt会多次就绪而et只就绪一次

EPOLL实现

  1. 数据结构

    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
    ##include <sys/queue.h>
    ##include "nty_tree.h"

    enum EPOLL_EVENTS {
    	EPOLLNONE 	= 0x0000,
    	EPOLLIN 	= 0x0001,
    	EPOLLPRI	= 0x0002,
    	EPOLLOUT	= 0x0004,
    	EPOLLRDNORM = 0x0040,
    	EPOLLRDBAND = 0x0080,
    	EPOLLWRNORM = 0x0100,
    	EPOLLWRBAND = 0x0200,
    	EPOLLMSG	= 0x0400,
    	EPOLLERR	= 0x0008,
    	EPOLLHUP 	= 0x0010,
    	EPOLLRDHUP 	= 0x2000,
    	EPOLLONESHOT = (1 << 30),
    	EPOLLET 	= (1 << 31)
    };

    ##define EPOLL_CTL_ADD	1
    ##define EPOLL_CTL_DEL	2
    ##define EPOLL_CTL_MOD	3

    typedef union epoll_data {
    	void *ptr;
    	int fd;
    	uint32_t u32;
    	uint64_t u64;
    } epoll_data_t;

    struct epoll_event {
    	uint32_t events;
    	epoll_data_t data;
    };

    /* 红黑树（就绪队列，共用）节点 */
    struct epitem {
    	RB_ENTRY(epitem) rbn;
    	LIST_ENTRY(epitem) rdlink;
    	int rdy; //exist in list 
    	int sockfd;
    	struct epoll_event event; 
    };

    static int sockfd_cmp(struct epitem *ep1, struct epitem *ep2) {
    	if (ep1->sockfd < ep2->sockfd) return -1;
    	else if (ep1->sockfd == ep2->sockfd) return 0;
    	return 1;
    }


    RB_HEAD(_epoll_rb_socket, epitem);
    RB_GENERATE_STATIC(_epoll_rb_socket, epitem, rbn, sockfd_cmp);

    typedef struct _epoll_rb_socket ep_rb_tree;

    /* 就绪队列根节点and红黑树根节点 */
    struct eventpoll {
    	int fd;
    	ep_rb_tree rbr;
    	int rbcnt;
    	LIST_HEAD( ,epitem) rdlist;
    	int rdnum;
    	int waiting;
    	// 根节点加锁
    	pthread_mutex_t mtx; //rbtree update
    	pthread_spinlock_t lock; //rdlist update
    	
    	pthread_cond_t cond; //block for event
    	pthread_mutex_t cdmtx; //mutex for cond
    	
    };

    int epoll_event_callback(struct eventpoll *ep, int sockid, uint32_t event);
    int nepoll_create(int size);
    int nepoll_ctl(int epfd, int op, int sockid, struct epoll_event *event);
    int nepoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

  2. api实现

    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
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165
    166
    167
    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200
    201
    202
    203
    204
    205
    206
    207
    208
    209
    210
    211
    212
    213
    214
    215
    216
    217
    218
    219
    220
    221
    222
    223
    224
    225
    226
    227
    228
    229
    230
    231
    232
    233
    // 3(应用层api) + 1(协议栈调用回调)

    // 由协议栈调用的回调
    int epoll_event_callback(struct eventpoll *ep, int sockid, uint32_t event) {
    	struct epitem tmp;
    	tmp.sockfd = sockid;
    	struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);
    	if (!epi) {
    		printf("rbtree not exist\n");
    		return -1;
    	}
    	if (epi->rdy) { // 已经在就绪队列里的只需要标志位或一下
    		epi->event.events |= event;
    		return 1;
    	} 
    	printf("epoll_event_callback --> %d\n", epi->sockfd);
    	pthread_spin_lock(&ep->lock);
    	epi->rdy = 1;
    	LIST_INSERT_HEAD(&ep->rdlist, epi, rdlink);
    	ep->rdnum ++;
    	pthread_spin_unlock(&ep->lock);

    	pthread_mutex_lock(&ep->cdmtx);

    	pthread_cond_signal(&ep->cond); // 通知epoll_wait
    	pthread_mutex_unlock(&ep->cdmtx);

    	return 0;
    }

    // eventpoll --> tcp_table->ep;
    int nepoll_create(int size) {
    	if (size <= 0) return -1;

    	// epfd --> struct eventpoll
    	int epfd = get_fd_frombitmap(); //tcp, udp	
    	struct eventpoll *ep = (struct eventpoll*)rte_malloc("eventpoll", sizeof(struct eventpoll), 0);
    	if (!ep) {
    		set_fd_frombitmap(epfd);
    		return -1;
    	}

    	struct ng_tcp_table *table = tcpInstance();
    	table->ep = ep;
    	
    	ep->fd = epfd;
    	ep->rbcnt = 0;
    	RB_INIT(&ep->rbr);
    	LIST_INIT(&ep->rdlist);

    	if (pthread_mutex_init(&ep->mtx, NULL)) {
    		free(ep);
    		set_fd_frombitmap(epfd);		
    		return -2;
    	}

    	if (pthread_mutex_init(&ep->cdmtx, NULL)) {
    		pthread_mutex_destroy(&ep->mtx);
    		free(ep);
    		set_fd_frombitmap(epfd);
    		return -2;
    	}

    	if (pthread_cond_init(&ep->cond, NULL)) {
    		pthread_mutex_destroy(&ep->cdmtx);
    		pthread_mutex_destroy(&ep->mtx);
    		free(ep);
    		set_fd_frombitmap(epfd);
    		return -2;
    	}

    	if (pthread_spin_init(&ep->lock, PTHREAD_PROCESS_SHARED)) {
    		pthread_cond_destroy(&ep->cond);
    		pthread_mutex_destroy(&ep->cdmtx);
    		pthread_mutex_destroy(&ep->mtx);
    		free(ep);

    		set_fd_frombitmap(epfd);
    		return -2;
    	}

    	return epfd;

    }


    int nepoll_ctl(int epfd, int op, int sockid, struct epoll_event *event) {
    	struct eventpoll *ep = (struct eventpoll*)get_hostinfo_fromfd(epfd);
    	if (!ep || (!event && op != EPOLL_CTL_DEL)) {
    		errno = -EINVAL;
    		return -1;
    	}

    	if (op == EPOLL_CTL_ADD) {
    		pthread_mutex_lock(&ep->mtx);

    		struct epitem tmp;
    		tmp.sockfd = sockid;
    		struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);
    		if (epi) { // 已经有了直接返回
    			pthread_mutex_unlock(&ep->mtx);
    			return -1;
    		}

    		epi = (struct epitem*)rte_malloc("epitem", sizeof(struct epitem), 0);
    		if (!epi) {
    			pthread_mutex_unlock(&ep->mtx);
    			rte_errno = -ENOMEM;
    			return -1;
    		}
    		
    		epi->sockfd = sockid;
    		memcpy(&epi->event, event, sizeof(struct epoll_event));

    		epi = RB_INSERT(_epoll_rb_socket, &ep->rbr, epi); // 插入

    		ep->rbcnt ++;
    		
    		pthread_mutex_unlock(&ep->mtx);

    	} else if (op == EPOLL_CTL_DEL) {

    		pthread_mutex_lock(&ep->mtx);

    		struct epitem tmp;
    		tmp.sockfd = sockid;
    		struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);
    		if (!epi) {
    			
    			pthread_mutex_unlock(&ep->mtx);
    			return -1;
    		}
    		
    		epi = RB_REMOVE(_epoll_rb_socket, &ep->rbr, epi); // 删除
    		if (!epi) {
    			
    			pthread_mutex_unlock(&ep->mtx);
    			return -1;
    		}

    		ep->rbcnt --;
    		rte_free(epi);
    		
    		pthread_mutex_unlock(&ep->mtx);

    	} else if (op == EPOLL_CTL_MOD) {

    		struct epitem tmp;
    		tmp.sockfd = sockid;
    		struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep->rbr, &tmp);
    		if (epi) { // 把事件复制进去就行了
    			epi->event.events = event->events;
    			epi->event.events |= EPOLLERR | EPOLLHUP;
    		} else {
    			rte_errno = -ENOENT;
    			return -1;
    		}
    	} 
    	return 0;
    }


    int nepoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout) {
    	struct eventpoll *ep = (struct eventpoll*)get_hostinfo_fromfd(epfd);;
    	if (!ep || !events || maxevents <= 0) {
    		rte_errno = -EINVAL;
    		return -1;
    	}
    	if (pthread_mutex_lock(&ep->cdmtx)) {
    		if (rte_errno == EDEADLK) {
    			printf("epoll lock blocked\n");
    		}
    	}

    	while (ep->rdnum == 0 && timeout != 0) {

    		ep->waiting = 1;
    		if (timeout > 0) { // timeout>0条件等待
    			struct timespec deadline;
    			clock_gettime(CLOCK_REALTIME, &deadline);
    			if (timeout >= 1000) {
    				int sec;
    				sec = timeout / 1000;
    				deadline.tv_sec += sec;
    				timeout -= sec * 1000;
    			}
    			deadline.tv_nsec += timeout * 1000000;
    			if (deadline.tv_nsec >= 1000000000) {
    				deadline.tv_sec++;
    				deadline.tv_nsec -= 1000000000;
    			}
    			int ret = pthread_cond_timedwait(&ep->cond, &ep->cdmtx, &deadline);
    			if (ret && ret != ETIMEDOUT) {
    				printf("pthread_cond_timewait\n");
    				
    				pthread_mutex_unlock(&ep->cdmtx);
    				
    				return -1;
    			}
    			timeout = 0;
    		} else if (timeout < 0) {
    			int ret = pthread_cond_wait(&ep->cond, &ep->cdmtx);
    			if (ret) {
    				printf("pthread_cond_wait\n");
    				pthread_mutex_unlock(&ep->cdmtx);
    				return -1;
    			}
    		}
    		ep->waiting = 0; 

    	}
    	// timeout == 0直接执行这里
    	pthread_mutex_unlock(&ep->cdmtx);
    	pthread_spin_lock(&ep->lock);

    	int cnt = 0;
    	int num = (ep->rdnum > maxevents ? maxevents : ep->rdnum);
    	int i = 0;	
    	while (num != 0 && !LIST_EMPTY(&ep->rdlist)) { //EPOLLET
    		struct epitem *epi = LIST_FIRST(&ep->rdlist);
    		LIST_REMOVE(epi, rdlink);
    		epi->rdy = 0;

    		// 从就绪队列中一个个搬到events里
    		memcpy(&events[i++], &epi->event, sizeof(struct epoll_event));
    		
    		num --;
    		cnt ++;
    		ep->rdnum --;
    	}
    	pthread_spin_unlock(&ep->lock);
    	return cnt;
    }

  3. 与协议栈交互

    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
    // 单个tcb表单个epoll模式
    struct ng_tcp_table {
    	int count;
    	//struct ng_tcp_stream *listener_set;	//
    ##if ENABLE_SINGLE_EPOLL 
    	struct eventpoll *ep; // single epoll
    ##endif
    	struct ng_tcp_stream *tcb_set;
    };

    // 在syn_recv状态进入establish状态时触发回调
    static int ng_tcp_handle_syn_rcvd(struct ng_tcp_stream *stream, struct rte_tcp_hdr *tcphdr) {
    	if (tcphdr->tcp_flags & RTE_TCP_ACK_FLAG) {
    		if (stream->status == NG_TCP_STATUS_SYN_RCVD) {
    			//......
    ##if ENABLE_SINGLE_EPOLL
    			struct ng_tcp_table *table = tcpInstance();
    			epoll_event_callback(table->ep, listener->fd, EPOLLIN);
    ##endif
    		}
    	}
    	return 0;
    }

    // 在establish状态收到数据、写入数据、fin位为1时触发回调
    static int ng_tcp_handle_established(struct ng_tcp_stream *stream, struct rte_tcp_hdr *tcphdr, int tcplen) {
    	if (tcphdr->tcp_flags & RTE_TCP_SYN_FLAG) {} 
    	if (tcphdr->tcp_flags & RTE_TCP_PSH_FLAG) {
    		ng_tcp_enqueue_recvbuffer(stream, tcphdr, tcplen);
    ##if ENABLE_SINGLE_EPOLL
    		struct ng_tcp_table *table = tcpInstance();
    		epoll_event_callback(table->ep, stream->fd, EPOLLIN);
    ##endif
    		uint8_t hdrlen = tcphdr->data_off >> 4;
    		int payloadlen = tcplen - hdrlen * 4;
    		stream->rcv_nxt = stream->rcv_nxt + payloadlen;
    		stream->snd_nxt = ntohl(tcphdr->recv_ack);
    		ng_tcp_send_ackpkt(stream, tcphdr);
    		// echo pkt
    	}
    	if (tcphdr->tcp_flags & RTE_TCP_ACK_FLAG) {}
    	if (tcphdr->tcp_flags & RTE_TCP_FIN_FLAG) {
    		stream->status = NG_TCP_STATUS_CLOSE_WAIT;
    		ng_tcp_enqueue_recvbuffer(stream, tcphdr, tcphdr->data_off >> 4);
    ##if ENABLE_SINGLE_EPOLL
    		struct ng_tcp_table *table = tcpInstance();
    		epoll_event_callback(table->ep, stream->fd, EPOLLIN);
    ##endif
    		// send ack ptk
    		stream->rcv_nxt = stream->rcv_nxt + 1;
    		stream->snd_nxt = ntohl(tcphdr->recv_ack);
    		ng_tcp_send_ackpkt(stream, tcphdr);

    ##if ENABLE_SINGLE_EPOLL
    		// 如果关注EPOLLOUT事件，那么这里进行回调
    ##endif
    	}
    	return 0;
    }

  4. tcp server调用api

    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
    static int tcp_server_entry(__attribute__((unused))  void *arg)  {
    	int listenfd = nsocket(AF_INET, SOCK_STREAM, 0);
    	if (listenfd == -1) {
    		return -1;
    	}
    	struct sockaddr_in servaddr;
    	memset(&servaddr, 0, sizeof(struct sockaddr));
    	servaddr.sin_family = AF_INET;
    	servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    	servaddr.sin_port = htons(9999);
    	nbind(listenfd, (struct sockaddr*)&servaddr, sizeof(servaddr));
    	nlisten(listenfd, 10);
    	printf("nlisten 9999 --> \n");
    	int epfd = nepoll_create(1); // event poll
    	struct epoll_event ev, events[128];
    	ev.events = EPOLLIN;
    	ev.data.fd = listenfd;
    	nepoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev);
    	char buff[BUFFER_SIZE] = {0};
    	while(1) {
    		int nready = nepoll_wait(epfd, events, 128, 5);
    		if (nready < 0) continue;
    		int i = 0;
    		for (i = 0;i < nready;i ++) {
    			if (listenfd == events[i].data.fd) {
    				struct sockaddr_in client;
    				socklen_t len = sizeof(client);
    				int connfd = naccept(listenfd, (struct sockaddr*)&client, &len);
    				struct epoll_event ev;
    				ev.events = EPOLLIN;
    				ev.data.fd = connfd;
    				nepoll_ctl(epfd, EPOLL_CTL_ADD, connfd, &ev);
    			} else { // clientfd
    				int connfd = events[i].data.fd;
    				int n = nrecv(connfd, buff, BUFFER_SIZE, 0); //block
    				if (n > 0) {
    					printf("recv: %s\n", buff);
    					nsend(connfd, buff, n, 0);
    				} else {
    					printf("recv: ret %d\n", n);
    					nepoll_ctl(epfd, EPOLL_CTL_DEL, connfd, NULL);
    					nclose(connfd);
    				} 
    			}
    		}
    	}
    }
2024-02-23 该篇文章被 Cleofwine 归为分类: 服务端

上一篇：后端技术_io_uring与liburing 下一篇：后端技术_应用层posix api具体实现
© 2023-2026 Cleofwine / Email / GitHub / 浙ICP备2024136127号
🌙赠我弯弯一枚月，也赠予我晚星⭐ 本站由Hexo驱动｜使用Hexo-theme-A4主题