Mercurial > hg > nginx

comparison src/event/ngx_event_udp.c @ 7285:88a624c9b491

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Events: moved ngx_recvmsg() to new file src/event/ngx_event_udp.c.
author Roman Arutyunyan <arut@nginx.com>
date Fri, 01 Jun 2018 16:55:49 +0300
parents
children d27aa9060c95
comparison
equal deleted inserted replaced
7284:52aacc8ddcc5 7285:88a624c9b491
1
2 /*
3 * Copyright (C) Roman Arutyunyan
4 * Copyright (C) Nginx, Inc.
5 */
6
7
8 #include <ngx_config.h>
9 #include <ngx_core.h>
10 #include <ngx_event.h>
11
12
13 #if !(NGX_WIN32)
14
15 static void ngx_close_accepted_udp_connection(ngx_connection_t *c);
16
17
18 void
19 ngx_event_recvmsg(ngx_event_t *ev)
20 {
21 ssize_t n;
22 ngx_log_t *log;
23 ngx_err_t err;
24 socklen_t socklen, local_socklen;
25 ngx_event_t *rev, *wev;
26 struct iovec iov[1];
27 struct msghdr msg;
28 ngx_sockaddr_t sa, lsa;
29 struct sockaddr *sockaddr, *local_sockaddr;
30 ngx_listening_t *ls;
31 ngx_event_conf_t *ecf;
32 ngx_connection_t *c, *lc;
33 static u_char buffer[65535];
34
35 #if (NGX_HAVE_MSGHDR_MSG_CONTROL)
36
37 #if (NGX_HAVE_IP_RECVDSTADDR)
38 u_char msg_control[CMSG_SPACE(sizeof(struct in_addr))];
39 #elif (NGX_HAVE_IP_PKTINFO)
40 u_char msg_control[CMSG_SPACE(sizeof(struct in_pktinfo))];
41 #endif
42
43 #if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO)
44 u_char msg_control6[CMSG_SPACE(sizeof(struct in6_pktinfo))];
45 #endif
46
47 #endif
48
49 if (ev->timedout) {
50 if (ngx_enable_accept_events((ngx_cycle_t *) ngx_cycle) != NGX_OK) {
51 return;
52 }
53
54 ev->timedout = 0;
55 }
56
57 ecf = ngx_event_get_conf(ngx_cycle->conf_ctx, ngx_event_core_module);
58
59 if (!(ngx_event_flags & NGX_USE_KQUEUE_EVENT)) {
60 ev->available = ecf->multi_accept;
61 }
62
63 lc = ev->data;
64 ls = lc->listening;
65 ev->ready = 0;
66
67 ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
68 "recvmsg on %V, ready: %d", &ls->addr_text, ev->available);
69
70 do {
71 ngx_memzero(&msg, sizeof(struct msghdr));
72
73 iov[0].iov_base = (void *) buffer;
74 iov[0].iov_len = sizeof(buffer);
75
76 msg.msg_name = &sa;
77 msg.msg_namelen = sizeof(ngx_sockaddr_t);
78 msg.msg_iov = iov;
79 msg.msg_iovlen = 1;
80
81 #if (NGX_HAVE_MSGHDR_MSG_CONTROL)
82
83 if (ls->wildcard) {
84
85 #if (NGX_HAVE_IP_RECVDSTADDR || NGX_HAVE_IP_PKTINFO)
86 if (ls->sockaddr->sa_family == AF_INET) {
87 msg.msg_control = &msg_control;
88 msg.msg_controllen = sizeof(msg_control);
89 }
90 #endif
91
92 #if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO)
93 if (ls->sockaddr->sa_family == AF_INET6) {
94 msg.msg_control = &msg_control6;
95 msg.msg_controllen = sizeof(msg_control6);
96 }
97 #endif
98 }
99
100 #endif
101
102 n = recvmsg(lc->fd, &msg, 0);
103
104 if (n == -1) {
105 err = ngx_socket_errno;
106
107 if (err == NGX_EAGAIN) {
108 ngx_log_debug0(NGX_LOG_DEBUG_EVENT, ev->log, err,
109 "recvmsg() not ready");
110 return;
111 }
112
113 ngx_log_error(NGX_LOG_ALERT, ev->log, err, "recvmsg() failed");
114
115 return;
116 }
117
118 #if (NGX_HAVE_MSGHDR_MSG_CONTROL)
119 if (msg.msg_flags & (MSG_TRUNC|MSG_CTRUNC)) {
120 ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
121 "recvmsg() truncated data");
122 continue;
123 }
124 #endif
125
126 sockaddr = msg.msg_name;
127 socklen = msg.msg_namelen;
128
129 if (socklen > (socklen_t) sizeof(ngx_sockaddr_t)) {
130 socklen = sizeof(ngx_sockaddr_t);
131 }
132
133 if (socklen == 0) {
134
135 /*
136 * on Linux recvmsg() returns zero msg_namelen
137 * when receiving packets from unbound AF_UNIX sockets
138 */
139
140 socklen = sizeof(struct sockaddr);
141 ngx_memzero(&sa, sizeof(struct sockaddr));
142 sa.sockaddr.sa_family = ls->sockaddr->sa_family;
143 }
144
145 local_sockaddr = ls->sockaddr;
146 local_socklen = ls->socklen;
147
148 #if (NGX_HAVE_MSGHDR_MSG_CONTROL)
149
150 if (ls->wildcard) {
151 struct cmsghdr *cmsg;
152
153 ngx_memcpy(&lsa, local_sockaddr, local_socklen);
154 local_sockaddr = &lsa.sockaddr;
155
156 for (cmsg = CMSG_FIRSTHDR(&msg);
157 cmsg != NULL;
158 cmsg = CMSG_NXTHDR(&msg, cmsg))
159 {
160
161 #if (NGX_HAVE_IP_RECVDSTADDR)
162
163 if (cmsg->cmsg_level == IPPROTO_IP
164 && cmsg->cmsg_type == IP_RECVDSTADDR
165 && local_sockaddr->sa_family == AF_INET)
166 {
167 struct in_addr *addr;
168 struct sockaddr_in *sin;
169
170 addr = (struct in_addr *) CMSG_DATA(cmsg);
171 sin = (struct sockaddr_in *) local_sockaddr;
172 sin->sin_addr = *addr;
173
174 break;
175 }
176
177 #elif (NGX_HAVE_IP_PKTINFO)
178
179 if (cmsg->cmsg_level == IPPROTO_IP
180 && cmsg->cmsg_type == IP_PKTINFO
181 && local_sockaddr->sa_family == AF_INET)
182 {
183 struct in_pktinfo *pkt;
184 struct sockaddr_in *sin;
185
186 pkt = (struct in_pktinfo *) CMSG_DATA(cmsg);
187 sin = (struct sockaddr_in *) local_sockaddr;
188 sin->sin_addr = pkt->ipi_addr;
189
190 break;
191 }
192
193 #endif
194
195 #if (NGX_HAVE_INET6 && NGX_HAVE_IPV6_RECVPKTINFO)
196
197 if (cmsg->cmsg_level == IPPROTO_IPV6
198 && cmsg->cmsg_type == IPV6_PKTINFO
199 && local_sockaddr->sa_family == AF_INET6)
200 {
201 struct in6_pktinfo *pkt6;
202 struct sockaddr_in6 *sin6;
203
204 pkt6 = (struct in6_pktinfo *) CMSG_DATA(cmsg);
205 sin6 = (struct sockaddr_in6 *) local_sockaddr;
206 sin6->sin6_addr = pkt6->ipi6_addr;
207
208 break;
209 }
210
211 #endif
212
213 }
214 }
215
216 #endif
217
218 #if (NGX_STAT_STUB)
219 (void) ngx_atomic_fetch_add(ngx_stat_accepted, 1);
220 #endif
221
222 ngx_accept_disabled = ngx_cycle->connection_n / 8
223 - ngx_cycle->free_connection_n;
224
225 c = ngx_get_connection(lc->fd, ev->log);
226 if (c == NULL) {
227 return;
228 }
229
230 c->shared = 1;
231 c->type = SOCK_DGRAM;
232 c->socklen = socklen;
233
234 #if (NGX_STAT_STUB)
235 (void) ngx_atomic_fetch_add(ngx_stat_active, 1);
236 #endif
237
238 c->pool = ngx_create_pool(ls->pool_size, ev->log);
239 if (c->pool == NULL) {
240 ngx_close_accepted_udp_connection(c);
241 return;
242 }
243
244 c->sockaddr = ngx_palloc(c->pool, socklen);
245 if (c->sockaddr == NULL) {
246 ngx_close_accepted_udp_connection(c);
247 return;
248 }
249
250 ngx_memcpy(c->sockaddr, sockaddr, socklen);
251
252 log = ngx_palloc(c->pool, sizeof(ngx_log_t));
253 if (log == NULL) {
254 ngx_close_accepted_udp_connection(c);
255 return;
256 }
257
258 *log = ls->log;
259
260 c->send = ngx_udp_send;
261 c->send_chain = ngx_udp_send_chain;
262
263 c->log = log;
264 c->pool->log = log;
265 c->listening = ls;
266
267 if (local_sockaddr == &lsa.sockaddr) {
268 local_sockaddr = ngx_palloc(c->pool, local_socklen);
269 if (local_sockaddr == NULL) {
270 ngx_close_accepted_udp_connection(c);
271 return;
272 }
273
274 ngx_memcpy(local_sockaddr, &lsa, local_socklen);
275 }
276
277 c->local_sockaddr = local_sockaddr;
278 c->local_socklen = local_socklen;
279
280 c->buffer = ngx_create_temp_buf(c->pool, n);
281 if (c->buffer == NULL) {
282 ngx_close_accepted_udp_connection(c);
283 return;
284 }
285
286 c->buffer->last = ngx_cpymem(c->buffer->last, buffer, n);
287
288 rev = c->read;
289 wev = c->write;
290
291 wev->ready = 1;
292
293 rev->log = log;
294 wev->log = log;
295
296 /*
297 * TODO: MT: - ngx_atomic_fetch_add()
298 * or protection by critical section or light mutex
299 *
300 * TODO: MP: - allocated in a shared memory
301 * - ngx_atomic_fetch_add()
302 * or protection by critical section or light mutex
303 */
304
305 c->number = ngx_atomic_fetch_add(ngx_connection_counter, 1);
306
307 #if (NGX_STAT_STUB)
308 (void) ngx_atomic_fetch_add(ngx_stat_handled, 1);
309 #endif
310
311 if (ls->addr_ntop) {
312 c->addr_text.data = ngx_pnalloc(c->pool, ls->addr_text_max_len);
313 if (c->addr_text.data == NULL) {
314 ngx_close_accepted_udp_connection(c);
315 return;
316 }
317
318 c->addr_text.len = ngx_sock_ntop(c->sockaddr, c->socklen,
319 c->addr_text.data,
320 ls->addr_text_max_len, 0);
321 if (c->addr_text.len == 0) {
322 ngx_close_accepted_udp_connection(c);
323 return;
324 }
325 }
326
327 #if (NGX_DEBUG)
328 {
329 ngx_str_t addr;
330 u_char text[NGX_SOCKADDR_STRLEN];
331
332 ngx_debug_accepted_connection(ecf, c);
333
334 if (log->log_level & NGX_LOG_DEBUG_EVENT) {
335 addr.data = text;
336 addr.len = ngx_sock_ntop(c->sockaddr, c->socklen, text,
337 NGX_SOCKADDR_STRLEN, 1);
338
339 ngx_log_debug4(NGX_LOG_DEBUG_EVENT, log, 0,
340 "*%uA recvmsg: %V fd:%d n:%z",
341 c->number, &addr, c->fd, n);
342 }
343
344 }
345 #endif
346
347 log->data = NULL;
348 log->handler = NULL;
349
350 ls->handler(c);
351
352 if (ngx_event_flags & NGX_USE_KQUEUE_EVENT) {
353 ev->available -= n;
354 }
355
356 } while (ev->available);
357 }
358
359
360 static void
361 ngx_close_accepted_udp_connection(ngx_connection_t *c)
362 {
363 ngx_free_connection(c);
364
365 c->fd = (ngx_socket_t) -1;
366
367 if (c->pool) {
368 ngx_destroy_pool(c->pool);
369 }
370
371 #if (NGX_STAT_STUB)
372 (void) ngx_atomic_fetch_add(ngx_stat_active, -1);
373 #endif
374 }
375
376 #endif