-
Notifications
You must be signed in to change notification settings - Fork 1
/
close_inactive_connection_tw_timer.c
232 lines (230 loc) · 7.62 KB
/
close_inactive_connection_tw_timer.c
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
#include "tw_timer.h"
#include <arpa/inet.h>
#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#define FD_LIMIT 65535
#define MAX_EVENT_NUMBER 1024
#define TIMEOUT 15
static int pipefd[2];
/*利用代码清单11-2中的升序链表来管理定时器*/
static time_wheel tw;
static int epollfd = 0;
int setnonblocking(int fd)
{
int old_option = fcntl(fd, F_GETFL);
int new_option = old_option | O_NONBLOCK;
fcntl(fd, F_SETFL, new_option);
return old_option;
}
void addfd(int epollfd, int fd)
{
struct epoll_event event;
event.data.fd = fd;
event.events = EPOLLIN | EPOLLET;
epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);
setnonblocking(fd);
}
void sig_handler(int sig)
{
int save_errno = errno;
int msg = sig;
send(pipefd[1], (char *)&msg, 1, 0);
errno = save_errno;
}
void addsig(int sig)
{
struct sigaction sa;
memset(&sa, '\0', sizeof(sa));
sa.sa_handler = sig_handler;
sa.sa_flags |= SA_RESTART;
sigfillset(&sa.sa_mask);
assert(sigaction(sig, &sa, NULL) != -1);
}
void timer_handler()
{
/*定时处理任务, 实际上就是调用tick函数*/
tw.tick();
/*因为一次alarm调用只会引起一次SIGALRM信号, 所以我们要重新定时,
* 以不断触发SIGALRM信号*/
alarm(SLOT_INTERVAL);
}
/*定时器回调函数, 它删除非活动连接socket上的注册事件, 并关闭之*/
void cb_func(client_data *user_data)
{
epoll_ctl(epollfd, EPOLL_CTL_DEL, user_data->sockfd, 0);
assert(user_data);
close(user_data->sockfd);
printf("close fd %d\n", user_data->sockfd);
}
int main(int argc, char *argv[])
{
if (argc <= 2)
{
printf("usage:%s ip_address port_number\n", basename(argv[0]));
return 1;
}
const char *ip = argv[1];
int port = atoi(argv[2]);
int ret = 0;
struct sockaddr_in address;
bzero(&address, sizeof(address));
address.sin_family = AF_INET;
inet_pton(AF_INET, ip, &address.sin_addr);
address.sin_port = htons(port);
int listenfd = socket(PF_INET, SOCK_STREAM, 0);
assert(listenfd >= 0);
ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address));
assert(ret != -1);
ret = listen(listenfd, 5);
assert(ret != -1);
epoll_event events[MAX_EVENT_NUMBER];
int epollfd = epoll_create(5);
assert(epollfd != -1);
addfd(epollfd, listenfd);
ret = socketpair(PF_UNIX, SOCK_STREAM, 0, pipefd);
assert(ret != -1);
setnonblocking(pipefd[1]);
addfd(epollfd, pipefd[0]);
/*设置信号处理函数*/
addsig(SIGALRM);
addsig(SIGTERM);
bool stop_server = false;
struct client_data *users = new struct client_data[FD_LIMIT];
bool timeout = false;
alarm(SLOT_INTERVAL); /*定时*/
while (!stop_server)
{
int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);
if ((number < 0) && (errno != EINTR))
{
printf("epoll failure\n");
break;
}
for (int i = 0; i < number; i++)
{
int sockfd = events[i].data.fd;
/*处理新到的客户连接*/
if (sockfd == listenfd)
{
struct sockaddr_in client_address;
socklen_t client_addrlength = sizeof(client_address);
int connfd = accept(listenfd, (struct sockaddr *)&client_address, &client_addrlength);
addfd(epollfd, connfd);
users[connfd].address = client_address;
users[connfd].sockfd = connfd;
/*创建定时器, 加入时间轮中,设置其回调函数与超时时间, 然后绑定定时器与用户数据*/
time_t cur = time(NULL);
tw_timer *timer = tw.add_timer(TIMEOUT);
timer->user_data = &users[connfd];
timer->cb_func = cb_func;
users[connfd].timer = timer;
}
/*处理信号*/
else if ((sockfd == pipefd[0]) && (events[i].events & EPOLLIN))
{
int sig;
char signals[1024];
ret = recv(pipefd[0], signals, sizeof(signals), 0);
if (ret == -1)
{
// handle the error
continue;
}
else if (ret == 0)
{
continue;
}
else
{
for (int i = 0; i < ret; ++i)
{
switch (signals[i])
{
case SIGALRM: {
/*用timeout变量标记有定时任务需要处理,
* 但不立即处理定时任务。这是因为定时任务的优先级不是很高,
* 我们优先处理其他更重要的任务*/
timeout = true;
break;
}
case SIGTERM: {
stop_server = true;
}
}
}
}
}
/*处理客户连接上接收到的数据*/
else if (events[i].events & EPOLLIN)
{
memset(users[sockfd].buf, '\0', BUFFER_SIZE);
ret = recv(sockfd, users[sockfd].buf, BUFFER_SIZE - 1, 0);
printf("get %d bytes of client data: %s from %d\n", ret, users[sockfd].buf, sockfd);
tw_timer *timer = users[sockfd].timer;
if (ret < 0)
{
/*如果发生读错误, 则关闭连接, 并移除其对应的定时器*/
if (errno != EAGAIN)
{
cb_func(&users[sockfd]);
if (timer)
{
tw.del_timer(timer);
}
}
}
else if (ret == 0)
{
/*如果对方已经关闭连接, 则我们也关闭连接, 并移除对应的定时器*/
cb_func(&users[sockfd]);
if (timer)
{
tw.del_timer(timer);
}
}
else
{
/*如果某个客户连接上有数据可读, 则我们要调整该连接对应的定时器,
* 以延迟该连接被关闭的时间*/
if (timer)
{
time_t cur = time(NULL);
tw_timer *new_timer = tw.add_timer(TIMEOUT);
new_timer->user_data = timer->user_data;
new_timer->cb_func = timer->cb_func;
new_timer->user_data->timer = new_timer;
printf("adjust timer once\n");
tw.del_timer(timer);
}
}
}
else
{
// others
}
}
/*最后处理定时事件, 因为I/O事件有更高的优先级。当然,
* 这样做将导致定时任务不能精确地按照预期的时间执行*/
if (timeout)
{
timer_handler();
timeout = false;
}
}
close(listenfd);
close(pipefd[1]);
close(pipefd[0]);
delete[] users;
return 0;
}