Mercurial > hg > nginx

comparison src/event/modules/ngx_kqueue_module.c @ 380:5ce6561246a5

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
nginx-0.0.7-2004-07-07-10:15:04 import
author Igor Sysoev <igor@sysoev.ru>
date Wed, 07 Jul 2004 06:15:04 +0000
parents 73688d5d7fc3
children 02a511569afb
comparison
equal deleted inserted replaced
379:73688d5d7fc3 380:5ce6561246a5
19 static ngx_int_t ngx_kqueue_init(ngx_cycle_t *cycle); 19 static ngx_int_t ngx_kqueue_init(ngx_cycle_t *cycle);
20 static void ngx_kqueue_done(ngx_cycle_t *cycle); 20 static void ngx_kqueue_done(ngx_cycle_t *cycle);
21 static ngx_int_t ngx_kqueue_add_event(ngx_event_t *ev, int event, u_int flags); 21 static ngx_int_t ngx_kqueue_add_event(ngx_event_t *ev, int event, u_int flags);
22 static ngx_int_t ngx_kqueue_del_event(ngx_event_t *ev, int event, u_int flags); 22 static ngx_int_t ngx_kqueue_del_event(ngx_event_t *ev, int event, u_int flags);
23 static ngx_int_t ngx_kqueue_set_event(ngx_event_t *ev, int filter, u_int flags); 23 static ngx_int_t ngx_kqueue_set_event(ngx_event_t *ev, int filter, u_int flags);
24 static ngx_int_t ngx_kqueue_process_changes(ngx_cycle_t *cycle, ngx_uint_t try);
24 static ngx_int_t ngx_kqueue_process_events(ngx_cycle_t *cycle); 25 static ngx_int_t ngx_kqueue_process_events(ngx_cycle_t *cycle);
25 static ngx_inline void ngx_kqueue_dump_event(ngx_log_t *log, 26 static ngx_inline void ngx_kqueue_dump_event(ngx_log_t *log,
26 struct kevent *kev); 27 struct kevent *kev);
27 28
28 static void *ngx_kqueue_create_conf(ngx_cycle_t *cycle); 29 static void *ngx_kqueue_create_conf(ngx_cycle_t *cycle);
29 static char *ngx_kqueue_init_conf(ngx_cycle_t *cycle, void *conf); 30 static char *ngx_kqueue_init_conf(ngx_cycle_t *cycle, void *conf);
30 31
31 32
32 int ngx_kqueue = -1; 33 int ngx_kqueue = -1;
33 34
34 static struct kevent *change_list, *event_list; 35 /*
36 * The "change_list" should be declared as ngx_thread_volatile.
37 * However, the use of the change_list is localized in kqueue functions and
38 * is protected by the mutex so even the "icc -ipo" should not build the code
39 * with the race condition. Thus we avoid the declaration to make a more
40 * readable code.
41 */
42
43 static struct kevent *change_list, *change_list0, *change_list1;
44 static struct kevent *event_list;
35 static int max_changes, nchanges, nevents; 45 static int max_changes, nchanges, nevents;
46
47 #if (NGX_THREADS)
48 static ngx_mutex_t *ngx_kqueue_mutex;
49 #endif
50
36 51
37 52
38 static ngx_str_t kqueue_name = ngx_string("kqueue"); 53 static ngx_str_t kqueue_name = ngx_string("kqueue");
39 54
40 static ngx_command_t ngx_kqueue_commands[] = { 55 static ngx_command_t ngx_kqueue_commands[] = {
67 ngx_kqueue_del_event, /* delete an event */ 82 ngx_kqueue_del_event, /* delete an event */
68 ngx_kqueue_add_event, /* enable an event */ 83 ngx_kqueue_add_event, /* enable an event */
69 ngx_kqueue_del_event, /* disable an event */ 84 ngx_kqueue_del_event, /* disable an event */
70 NULL, /* add an connection */ 85 NULL, /* add an connection */
71 NULL, /* delete an connection */ 86 NULL, /* delete an connection */
87 ngx_kqueue_process_changes, /* process the changes */
72 ngx_kqueue_process_events, /* process the events */ 88 ngx_kqueue_process_events, /* process the events */
73 ngx_kqueue_init, /* init the events */ 89 ngx_kqueue_init, /* init the events */
74 ngx_kqueue_done /* done the events */ 90 ngx_kqueue_done /* done the events */
75 } 91 }
76 92
80 NGX_MODULE, 96 NGX_MODULE,
81 &ngx_kqueue_module_ctx, /* module context */ 97 &ngx_kqueue_module_ctx, /* module context */
82 ngx_kqueue_commands, /* module directives */ 98 ngx_kqueue_commands, /* module directives */
83 NGX_EVENT_MODULE, /* module type */ 99 NGX_EVENT_MODULE, /* module type */
84 NULL, /* init module */ 100 NULL, /* init module */
85 NULL /* init child */ 101 NULL /* init process */
86 }; 102 };
87 103
88 104
89 105
90 static ngx_int_t ngx_kqueue_init(ngx_cycle_t *cycle) 106 static ngx_int_t ngx_kqueue_init(ngx_cycle_t *cycle)
100 if (ngx_kqueue == -1) { 116 if (ngx_kqueue == -1) {
101 ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno, 117 ngx_log_error(NGX_LOG_EMERG, cycle->log, ngx_errno,
102 "kqueue() failed"); 118 "kqueue() failed");
103 return NGX_ERROR; 119 return NGX_ERROR;
104 } 120 }
121
122 #if (NGX_THREADS)
123 if (!(ngx_kqueue_mutex = ngx_mutex_init(cycle->log, 0))) {
124 return NGX_ERROR;
125 }
126 #endif
105 } 127 }
106 128
107 if (max_changes < kcf->changes) { 129 if (max_changes < kcf->changes) {
108 if (nchanges) { 130 if (nchanges) {
109 ts.tv_sec = 0; 131 ts.tv_sec = 0;
115 return NGX_ERROR; 137 return NGX_ERROR;
116 } 138 }
117 nchanges = 0; 139 nchanges = 0;
118 } 140 }
119 141
120 if (change_list) { 142 if (change_list0) {
121 ngx_free(change_list); 143 ngx_free(change_list0);
122 } 144 }
123 145
124 change_list = ngx_alloc(kcf->changes * sizeof(struct kevent), 146 change_list0 = ngx_alloc(kcf->changes * sizeof(struct kevent),
125 cycle->log); 147 cycle->log);
126 if (change_list == NULL) { 148 if (change_list0 == NULL) {
127 return NGX_ERROR; 149 return NGX_ERROR;
128 } 150 }
151
152 if (change_list1) {
153 ngx_free(change_list1);
154 }
155
156 change_list1 = ngx_alloc(kcf->changes * sizeof(struct kevent),
157 cycle->log);
158 if (change_list1 == NULL) {
159 return NGX_ERROR;
160 }
161
162 change_list = change_list0;
129 } 163 }
130 164
131 max_changes = kcf->changes; 165 max_changes = kcf->changes;
132 166
133 if (nevents < kcf->events) { 167 if (nevents < kcf->events) {
134 if (event_list) { 168 if (event_list) {
135 ngx_free(event_list); 169 ngx_free(event_list);
136 } 170 }
137 171
138 event_list = ngx_alloc(kcf->events * sizeof(struct kevent), 172 event_list = ngx_alloc(kcf->events * sizeof(struct kevent), cycle->log);
139 cycle->log);
140 if (event_list == NULL) { 173 if (event_list == NULL) {
141 return NGX_ERROR; 174 return NGX_ERROR;
142 } 175 }
143 } 176 }
144 177
170 "kqueue close() failed"); 203 "kqueue close() failed");
171 } 204 }
172 205
173 ngx_kqueue = -1; 206 ngx_kqueue = -1;
174 207
175 ngx_free(change_list); 208 ngx_mutex_destroy(ngx_kqueue_mutex);
209
210 ngx_free(change_list1);
211 ngx_free(change_list0);
176 ngx_free(event_list); 212 ngx_free(event_list);
177 213
214 change_list1 = NULL;
215 change_list0 = NULL;
178 change_list = NULL; 216 change_list = NULL;
179 event_list = NULL; 217 event_list = NULL;
180 max_changes = 0; 218 max_changes = 0;
181 nchanges = 0; 219 nchanges = 0;
182 nevents = 0; 220 nevents = 0;
183 } 221 }
184 222
185 223
186 static ngx_int_t ngx_kqueue_add_event(ngx_event_t *ev, int event, u_int flags) 224 static ngx_int_t ngx_kqueue_add_event(ngx_event_t *ev, int event, u_int flags)
187 { 225 {
226 ngx_int_t rc;
188 ngx_event_t *e; 227 ngx_event_t *e;
189 ngx_connection_t *c; 228 ngx_connection_t *c;
190 229
191 ev->active = 1; 230 ev->active = 1;
192 ev->disabled = 0; 231 ev->disabled = 0;
193 ev->oneshot = (flags & NGX_ONESHOT_EVENT) ? 1 : 0; 232 ev->oneshot = (flags & NGX_ONESHOT_EVENT) ? 1 : 0;
194 233
195 if (ngx_thread_main() 234 if (ngx_mutex_lock(ngx_kqueue_mutex) == NGX_ERROR) {
196 && nchanges > 0 235 return NGX_ERROR;
236 }
237
238 if (nchanges > 0
197 && ev->index < (u_int) nchanges 239 && ev->index < (u_int) nchanges
198 && ((uintptr_t) change_list[ev->index].udata & (uintptr_t) ~1) 240 && ((uintptr_t) change_list[ev->index].udata & (uintptr_t) ~1)
199 == (uintptr_t) ev) 241 == (uintptr_t) ev)
200 { 242 {
201 if (change_list[ev->index].flags == EV_DISABLE) { 243 if (change_list[ev->index].flags == EV_DISABLE) {
213 e = (ngx_event_t *) change_list[nchanges].udata; 255 e = (ngx_event_t *) change_list[nchanges].udata;
214 change_list[ev->index] = change_list[nchanges]; 256 change_list[ev->index] = change_list[nchanges];
215 e->index = ev->index; 257 e->index = ev->index;
216 } 258 }
217 259
260 ngx_mutex_unlock(ngx_kqueue_mutex);
261
218 return NGX_OK; 262 return NGX_OK;
219 } 263 }
220 264
221 c = ev->data; 265 c = ev->data;
266
222 ngx_log_error(NGX_LOG_ALERT, ev->log, 0, 267 ngx_log_error(NGX_LOG_ALERT, ev->log, 0,
223 "previous event on #%d were not passed in kernel", c->fd); 268 "previous event on #%d were not passed in kernel", c->fd);
224 269
270 ngx_mutex_unlock(ngx_kqueue_mutex);
271
225 return NGX_ERROR; 272 return NGX_ERROR;
226 } 273 }
227 274
228 return ngx_kqueue_set_event(ev, event, EV_ADD|EV_ENABLE|flags); 275 rc = ngx_kqueue_set_event(ev, event, EV_ADD|EV_ENABLE|flags);
276
277 ngx_mutex_unlock(ngx_kqueue_mutex);
278
279 return rc;
229 } 280 }
230 281
231 282
232 static ngx_int_t ngx_kqueue_del_event(ngx_event_t *ev, int event, u_int flags) 283 static ngx_int_t ngx_kqueue_del_event(ngx_event_t *ev, int event, u_int flags)
233 { 284 {
285 ngx_int_t rc;
234 ngx_event_t *e; 286 ngx_event_t *e;
235 287
236 ev->active = 0; 288 ev->active = 0;
237 ev->disabled = 0; 289 ev->disabled = 0;
238 290
239 if (ngx_thread_main() 291 if (ngx_mutex_lock(ngx_kqueue_mutex) == NGX_ERROR) {
240 && nchanges > 0 292 return NGX_ERROR;
293 }
294
295 if (nchanges > 0
241 && ev->index < (u_int) nchanges 296 && ev->index < (u_int) nchanges
242 && ((uintptr_t) change_list[ev->index].udata & (uintptr_t) ~1) 297 && ((uintptr_t) change_list[ev->index].udata & (uintptr_t) ~1)
243 == (uintptr_t) ev) 298 == (uintptr_t) ev)
244 { 299 {
245 ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0, 300 ngx_log_debug2(NGX_LOG_DEBUG_EVENT, ev->log, 0,
252 e = (ngx_event_t *) change_list[nchanges].udata; 307 e = (ngx_event_t *) change_list[nchanges].udata;
253 change_list[ev->index] = change_list[nchanges]; 308 change_list[ev->index] = change_list[nchanges];
254 e->index = ev->index; 309 e->index = ev->index;
255 } 310 }
256 311
312 ngx_mutex_unlock(ngx_kqueue_mutex);
313
257 return NGX_OK; 314 return NGX_OK;
258 } 315 }
259 316
260 /* 317 /*
261 * when the file descriptor is closed the kqueue automatically deletes 318 * when the file descriptor is closed the kqueue automatically deletes
262 * its filters so we do not need to delete explicity the event 319 * its filters so we do not need to delete explicity the event
263 * before the closing the file descriptor. 320 * before the closing the file descriptor.
264 */ 321 */
265 322
266 if (flags & NGX_CLOSE_EVENT) { 323 if (flags & NGX_CLOSE_EVENT) {
324 ngx_mutex_unlock(ngx_kqueue_mutex);
267 return NGX_OK; 325 return NGX_OK;
268 } 326 }
269 327
270 if (flags & NGX_DISABLE_EVENT) { 328 if (flags & NGX_DISABLE_EVENT) {
271 ev->disabled = 1; 329 ev->disabled = 1;
272 } 330 }
273 331
274 return ngx_kqueue_set_event(ev, event, 332 rc = ngx_kqueue_set_event(ev, event,
275 flags & NGX_DISABLE_EVENT ? EV_DISABLE : EV_DELETE); 333 flags & NGX_DISABLE_EVENT ? EV_DISABLE : EV_DELETE);
334
335 ngx_mutex_unlock(ngx_kqueue_mutex);
336
337 return rc;
276 } 338 }
277 339
278 340
279 static ngx_int_t ngx_kqueue_set_event(ngx_event_t *ev, int filter, u_int flags) 341 static ngx_int_t ngx_kqueue_set_event(ngx_event_t *ev, int filter, u_int flags)
280 { 342 {
281 struct kevent *kev, kv; 343 struct kevent *kev;
282 struct timespec ts; 344 struct timespec ts;
283 ngx_connection_t *c; 345 ngx_connection_t *c;
284 346
285 c = ev->data; 347 c = ev->data;
286 348
287 ngx_log_debug3(NGX_LOG_DEBUG_EVENT, c->log, 0, 349 ngx_log_debug3(NGX_LOG_DEBUG_EVENT, c->log, 0,
288 "kevent set event: %d: ft:%d fl:%04X", 350 "kevent set event: %d: ft:%d fl:%04X",
289 c->fd, filter, flags); 351 c->fd, filter, flags);
290 352
291 if (ngx_thread_main() && nchanges >= max_changes) { 353 if (nchanges >= max_changes) {
292 ngx_log_error(NGX_LOG_WARN, ev->log, 0, 354 ngx_log_error(NGX_LOG_WARN, ev->log, 0,
293 "kqueue change list is filled up"); 355 "kqueue change list is filled up");
294 356
295 ts.tv_sec = 0; 357 ts.tv_sec = 0;
296 ts.tv_nsec = 0; 358 ts.tv_nsec = 0;
301 } 363 }
302 364
303 nchanges = 0; 365 nchanges = 0;
304 } 366 }
305 367
306 kev = ngx_thread_main() ? &change_list[nchanges] : &kv; 368 kev = &change_list[nchanges];
307 369
308 kev->ident = c->fd; 370 kev->ident = c->fd;
309 kev->filter = filter; 371 kev->filter = filter;
310 kev->flags = flags; 372 kev->flags = flags;
311 kev->udata = (void *) ((uintptr_t) ev | ev->instance); 373 kev->udata = (void *) ((uintptr_t) ev | ev->instance);
334 kev->fflags = 0; 396 kev->fflags = 0;
335 kev->data = 0; 397 kev->data = 0;
336 #endif 398 #endif
337 } 399 }
338 400
339 if (ngx_thread_main()) { 401 ev->index = nchanges;
340 ev->index = nchanges; 402 nchanges++;
341 nchanges++;
342
343 } else {
344 ts.tv_sec = 0;
345 ts.tv_nsec = 0;
346
347 if (kevent(ngx_kqueue, &kv, 1, NULL, 0, &ts) == -1) {
348 ngx_log_error(NGX_LOG_ALERT, ev->log, ngx_errno, "kevent() failed");
349 return NGX_ERROR;
350 }
351 }
352 403
353 return NGX_OK; 404 return NGX_OK;
354 } 405 }
355 406
356 407
357 static ngx_int_t ngx_kqueue_process_events(ngx_cycle_t *cycle) 408 static ngx_int_t ngx_kqueue_process_events(ngx_cycle_t *cycle)
358 { 409 {
359 int events; 410 int events, n;
360 ngx_int_t i, instance; 411 ngx_int_t i, instance;
361 ngx_uint_t lock, accept_lock, expire; 412 ngx_uint_t lock, accept_lock, expire;
362 ngx_err_t err; 413 ngx_err_t err;
363 ngx_msec_t timer; 414 ngx_msec_t timer;
364 ngx_event_t *ev; 415 ngx_event_t *ev;
417 expire = 0; 468 expire = 0;
418 } 469 }
419 } 470 }
420 } 471 }
421 472
473 if (ngx_threaded) {
474 if (ngx_kqueue_process_changes(cycle, 0) == NGX_ERROR) {
475 ngx_accept_mutex_unlock();
476 return NGX_ERROR;
477 }
478
479 n = 0;
480
481 } else {
482 n = nchanges;
483 nchanges = 0;
484 }
485
422 if (timer == NGX_TIMER_INFINITE) { 486 if (timer == NGX_TIMER_INFINITE) {
423 tp = NULL; 487 tp = NULL;
424 expire = 0; 488 expire = 0;
425 489
426 } else { 490 } else {
427 ts.tv_sec = timer / 1000; 491 ts.tv_sec = timer / 1000;
428 ts.tv_nsec = (timer % 1000) * 1000000; 492 ts.tv_nsec = (timer % 1000) * 1000000;
429 tp = &ts; 493 tp = &ts;
430 } 494 }
431 495
432 ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0, 496 ngx_log_debug2(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
433 "kevent timer: %d", timer); 497 "kevent timer: %d, changes: %d", timer, n);
434 498
435 events = kevent(ngx_kqueue, change_list, nchanges, event_list, nevents, tp); 499 events = kevent(ngx_kqueue, change_list, n, event_list, nevents, tp);
436 500
437 if (events == -1) { 501 if (events == -1) {
438 err = ngx_errno; 502 err = ngx_errno;
439 } else { 503 } else {
440 err = 0; 504 err = 0;
441 } 505 }
442
443 nchanges = 0;
444 506
445 ngx_gettimeofday(&tv); 507 ngx_gettimeofday(&tv);
446 ngx_time_update(tv.tv_sec); 508 ngx_time_update(tv.tv_sec);
447 509
448 ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0, 510 ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
624 ngx_event_process_posted(cycle); 686 ngx_event_process_posted(cycle);
625 } 687 }
626 } 688 }
627 689
628 return NGX_OK; 690 return NGX_OK;
691 }
692
693
694 static ngx_int_t ngx_kqueue_process_changes(ngx_cycle_t *cycle, ngx_uint_t try)
695 {
696 int n;
697 ngx_int_t rc;
698 ngx_err_t err;
699 struct timespec ts;
700 struct kevent *changes;
701
702 if (try) {
703 rc = ngx_mutex_trylock(ngx_kqueue_mutex);
704 if (rc != NGX_OK) {
705 return rc;
706 }
707
708 } else {
709 if (ngx_mutex_lock(ngx_kqueue_mutex) == NGX_ERROR) {
710 return NGX_ERROR;
711 }
712 }
713
714 if (nchanges == 0) {
715 ngx_mutex_unlock(ngx_kqueue_mutex);
716 return NGX_OK;
717 }
718
719 changes = (struct kevent *) change_list;
720 if (change_list == change_list0) {
721 change_list = change_list1;
722 } else {
723 change_list = change_list0;
724 }
725
726 n = nchanges;
727 nchanges = 0;
728
729 ts.tv_sec = 0;
730 ts.tv_nsec = 0;
731
732 ngx_log_debug1(NGX_LOG_DEBUG_EVENT, cycle->log, 0,
733 "kevent changes: %d", n);
734
735 if (kevent(ngx_kqueue, changes, n, NULL, 0, &ts) == -1) {
736 err = ngx_errno;
737 ngx_log_error((err == NGX_EINTR) ? NGX_LOG_INFO : NGX_LOG_ALERT,
738 cycle->log, err, "kevent() failed");
739 rc = NGX_ERROR;
740
741 } else {
742 rc = NGX_OK;
743 }
744
745 ngx_mutex_unlock(ngx_kqueue_mutex);
746
747 return rc;
629 } 748 }
630 749
631 750
632 static ngx_inline void ngx_kqueue_dump_event(ngx_log_t *log, struct kevent *kev) 751 static ngx_inline void ngx_kqueue_dump_event(ngx_log_t *log, struct kevent *kev)
633 { 752 {