1
2 /*
3 * Copyright (C) Igor Sysoev
4 */
5
6
7 #include <ngx_config.h>
8 #include <ngx_core.h>
9
10 /*
11 * The threads implementation uses the rfork(RFPROC|RFTHREAD|RFMEM) syscall
12 * to create threads. All threads use the stacks of the same size mmap()ed
13 * below the main stack. Thus the current thread id is determinated via
14 * the stack pointer value.
15 *
16 * The mutex implementation uses the ngx_atomic_cmp_set() operation
17 * to acquire a mutex and the SysV semaphore to wait on a mutex and to wake up
18 * the waiting threads. The light mutex does not use semaphore, so after
19 * spinning in the lock the thread calls sched_yield(). However the light
20 * mutecies are intended to be used with the "trylock" operation only.
21 * The SysV semop() is a cheap syscall, particularly if it has little sembuf's
22 * and does not use SEM_UNDO.
23 *
24 * The condition variable implementation uses the signal #64.
25 * The signal handler is SIG_IGN so the kill() is a cheap syscall.
26 * The thread waits a signal in kevent(). The use of the EVFILT_SIGNAL
27 * is safe since FreeBSD 4.10-STABLE.
28 *
29 * This threads implementation currently works on i386 (486+) and amd64
30 * platforms only.
31 */
32
33
34 char *ngx_freebsd_kern_usrstack;
35 size_t ngx_thread_stack_size;
36
37
38 static size_t rz_size;
39 static size_t usable_stack_size;
40 static char *last_stack;
41
42 static ngx_uint_t nthreads;
43 static ngx_uint_t max_threads;
44
45 static ngx_uint_t nkeys;
46 static ngx_tid_t *tids; /* the threads tids array */
47 void **ngx_tls; /* the threads tls's array */
48
49 /* the thread-safe libc errno */
50
51 static int errno0; /* the main thread's errno */
52 static int *errnos; /* the threads errno's array */
53
54 int *
55 __error()
56 {
57 int tid;
58
59 tid = ngx_gettid();
60
61 return tid ? &errnos[tid - 1] : &errno0;
62 }
63
64
65 /*
66 * __isthreaded enables the spinlocks in some libc functions, i.e. in malloc()
67 * and some other places. Nevertheless we protect our malloc()/free() calls
68 * by own mutex that is more efficient than the spinlock.
69 *
70 * _spinlock() is a weak referenced stub in src/lib/libc/gen/_spinlock_stub.c
71 * that does nothing.
72 */
73
74 extern int __isthreaded;
75
76 void
77 _spinlock(ngx_atomic_t *lock)
78 {
79 ngx_int_t tries;
80
81 tries = 0;
82
83 for ( ;; ) {
84
85 if (*lock) {
86 if (ngx_ncpu > 1 && tries++ < 1000) {
87 continue;
88 }
89
90 sched_yield();
91 tries = 0;
92
93 } else {
94 if (ngx_atomic_cmp_set(lock, 0, 1)) {
95 return;
96 }
97 }
98 }
99 }
100
101
102 /*
103 * Before FreeBSD 5.1 _spinunlock() is a simple #define in
104 * src/lib/libc/include/spinlock.h that zeroes lock.
105 *
106 * Since FreeBSD 5.1 _spinunlock() is a weak referenced stub in
107 * src/lib/libc/gen/_spinlock_stub.c that does nothing.
108 */
109
110 #ifndef _spinunlock
111
112 void
113 _spinunlock(ngx_atomic_t *lock)
114 {
115 *lock = 0;
116 }
117
118 #endif
119
120
121 ngx_err_t
122 ngx_create_thread(ngx_tid_t *tid, ngx_thread_value_t (*func)(void *arg),
123 void *arg, ngx_log_t *log)
124 {
125 ngx_pid_t id;
126 ngx_err_t err;
127 char *stack, *stack_top;
128
129 if (nthreads >= max_threads) {
130 ngx_log_error(NGX_LOG_CRIT, log, 0,
131 "no more than %ui threads can be created", max_threads);
132 return NGX_ERROR;
133 }
134
135 last_stack -= ngx_thread_stack_size;
136
137 stack = mmap(last_stack, usable_stack_size, PROT_READ|PROT_WRITE,
138 MAP_STACK, -1, 0);
139
140 if (stack == MAP_FAILED) {
141 ngx_log_error(NGX_LOG_ALERT, log, ngx_errno,
142 "mmap(%p:%uz, MAP_STACK) thread stack failed",
143 last_stack, usable_stack_size);
144 return NGX_ERROR;
145 }
146
147 if (stack != last_stack) {
148 ngx_log_error(NGX_LOG_ALERT, log, 0,
149 "stack %p address was changed to %p", last_stack, stack);
150 return NGX_ERROR;
151 }
152
153 stack_top = stack + usable_stack_size;
154
155 ngx_log_debug2(NGX_LOG_DEBUG_CORE, log, 0,
156 "thread stack: %p-%p", stack, stack_top);
157
158 ngx_set_errno(0);
159
160 id = rfork_thread(RFPROC|RFTHREAD|RFMEM, stack_top,
161 (ngx_rfork_thread_func_pt) func, arg);
162
163 err = ngx_errno;
164
165 if (id == -1) {
166 ngx_log_error(NGX_LOG_ALERT, log, err, "rfork() failed");
167
168 } else {
169 *tid = id;
170 nthreads = (ngx_freebsd_kern_usrstack - stack_top)
171 / ngx_thread_stack_size;
172 tids[nthreads] = id;
173
174 ngx_log_debug1(NGX_LOG_DEBUG_CORE, log, 0, "rfork()ed thread: %P", id);
175 }
176
177 return err;
178 }
179
180
181 ngx_int_t
182 ngx_init_threads(int n, size_t size, ngx_cycle_t *cycle)
183 {
184 char *red_zone, *zone;
185 size_t len;
186 ngx_int_t i;
187 struct sigaction sa;
188
189 max_threads = n + 1;
190
191 for (i = 0; i < n; i++) {
192 ngx_memzero(&sa, sizeof(struct sigaction));
193 sa.sa_handler = SIG_IGN;
194 sigemptyset(&sa.sa_mask);
195 if (sigaction(NGX_CV_SIGNAL, &sa, NULL) == -1) {
196 ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
197 "sigaction(%d, SIG_IGN) failed", NGX_CV_SIGNAL);
198 return NGX_ERROR;
199 }
200 }
201
202 len = sizeof(ngx_freebsd_kern_usrstack);
203 if (sysctlbyname("kern.usrstack", &ngx_freebsd_kern_usrstack, &len,
204 NULL, 0) == -1)
205 {
206 ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
207 "sysctlbyname(kern.usrstack) failed");
208 return NGX_ERROR;
209 }
210
211 /* the main thread stack red zone */
212 rz_size = ngx_pagesize;
213 red_zone = ngx_freebsd_kern_usrstack - (size + rz_size);
214
215 ngx_log_debug2(NGX_LOG_DEBUG_CORE, cycle->log, 0,
216 "usrstack: %p red zone: %p",
217 ngx_freebsd_kern_usrstack, red_zone);
218
219 zone = mmap(red_zone, rz_size, PROT_NONE, MAP_ANON, -1, 0);
220 if (zone == MAP_FAILED) {
221 ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
222 "mmap(%p:%uz, PROT_NONE, MAP_ANON) red zone failed",
223 red_zone, rz_size);
224 return NGX_ERROR;
225 }
226
227 if (zone != red_zone) {
228 ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
229 "red zone %p address was changed to %p", red_zone, zone);
230 return NGX_ERROR;
231 }
232
233 /* create the thread errno' array */
234
235 errnos = ngx_calloc(n * sizeof(int), cycle->log);
236 if (errnos == NULL) {
237 return NGX_ERROR;
238 }
239
240 /* create the thread tids array */
241
242 tids = ngx_calloc((n + 1) * sizeof(ngx_tid_t), cycle->log);
243 if (tids == NULL) {
244 return NGX_ERROR;
245 }
246
247 tids[0] = ngx_pid;
248
249 /* create the thread tls' array */
250
251 ngx_tls = ngx_calloc(NGX_THREAD_KEYS_MAX * (n + 1) * sizeof(void *),
252 cycle->log);
253 if (ngx_tls == NULL) {
254 return NGX_ERROR;
255 }
256
257 nthreads = 1;
258
259 last_stack = zone + rz_size;
260 usable_stack_size = size;
261 ngx_thread_stack_size = size + rz_size;
262
263 /* allow the spinlock in libc malloc() */
264 __isthreaded = 1;
265
266 ngx_threaded = 1;
267
268 return NGX_OK;
269 }
270
271
272 ngx_tid_t
273 ngx_thread_self()
274 {
275 ngx_int_t tid;
276
277 tid = ngx_gettid();
278
279 if (tids == NULL) {
280 return ngx_pid;
281 }
282
283 return tids[tid];
284 }
285
286
287 ngx_err_t
288 ngx_thread_key_create(ngx_tls_key_t *key)
289 {
290 if (nkeys >= NGX_THREAD_KEYS_MAX) {
291 return NGX_ENOMEM;
292 }
293
294 *key = nkeys++;
295
296 return 0;
297 }
298
299
300 ngx_err_t
301 ngx_thread_set_tls(ngx_tls_key_t key, void *value)
302 {
303 if (key >= NGX_THREAD_KEYS_MAX) {
304 return NGX_EINVAL;
305 }
306
307 ngx_tls[key * NGX_THREAD_KEYS_MAX + ngx_gettid()] = value;
308 return 0;
309 }
310
311
312 ngx_mutex_t *
313 ngx_mutex_init(ngx_log_t *log, ngx_uint_t flags)
314 {
315 ngx_mutex_t *m;
316 union semun op;
317
318 m = ngx_alloc(sizeof(ngx_mutex_t), log);
319 if (m == NULL) {
320 return NULL;
321 }
322
323 m->lock = 0;
324 m->log = log;
325
326 if (flags & NGX_MUTEX_LIGHT) {
327 m->semid = -1;
328 return m;
329 }
330
331 m->semid = semget(IPC_PRIVATE, 1, SEM_R|SEM_A);
332 if (m->semid == -1) {
333 ngx_log_error(NGX_LOG_ALERT, log, ngx_errno, "semget() failed");
334 return NULL;
335 }
336
337 op.val = 0;
338
339 if (semctl(m->semid, 0, SETVAL, op) == -1) {
340 ngx_log_error(NGX_LOG_ALERT, log, ngx_errno, "semctl(SETVAL) failed");
341
342 if (semctl(m->semid, 0, IPC_RMID) == -1) {
343 ngx_log_error(NGX_LOG_ALERT, log, ngx_errno,
344 "semctl(IPC_RMID) failed");
345 }
346
347 return NULL;
348 }
349
350 return m;
351 }
352
353
354 void
355 ngx_mutex_destroy(ngx_mutex_t *m)
356 {
357 if (semctl(m->semid, 0, IPC_RMID) == -1) {
358 ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
359 "semctl(IPC_RMID) failed");
360 }
361
362 ngx_free((void *) m);
363 }
364
365
366 ngx_int_t
367 ngx_mutex_dolock(ngx_mutex_t *m, ngx_int_t try)
368 {
369 uint32_t lock, old;
370 ngx_uint_t tries;
371 struct sembuf op;
372
373 if (!ngx_threaded) {
374 return NGX_OK;
375 }
376
377 #if (NGX_DEBUG)
378 if (try) {
379 ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
380 "try lock mutex %p lock:%XD", m, m->lock);
381 } else {
382 ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
383 "lock mutex %p lock:%XD", m, m->lock);
384 }
385 #endif
386
387 old = m->lock;
388 tries = 0;
389
390 for ( ;; ) {
391 if (old & NGX_MUTEX_LOCK_BUSY) {
392
393 if (try) {
394 return NGX_AGAIN;
395 }
396
397 if (ngx_ncpu > 1 && tries++ < 1000) {
398
399 /* the spinlock is used only on the SMP system */
400
401 old = m->lock;
402 continue;
403 }
404
405 if (m->semid == -1) {
406 sched_yield();
407
408 tries = 0;
409 old = m->lock;
410 continue;
411 }
412
413 ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
414 "mutex %p lock:%XD", m, m->lock);
415
416 /*
417 * The mutex is locked so we increase a number
418 * of the threads that are waiting on the mutex
419 */
420
421 lock = old + 1;
422
423 if ((lock & ~NGX_MUTEX_LOCK_BUSY) > nthreads) {
424 ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
425 "%D threads wait for mutex %p, "
426 "while only %ui threads are available",
427 lock & ~NGX_MUTEX_LOCK_BUSY, m, nthreads);
428 ngx_abort();
429 }
430
431 if (ngx_atomic_cmp_set(&m->lock, old, lock)) {
432
433 ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
434 "wait mutex %p lock:%XD", m, m->lock);
435
436 /*
437 * The number of the waiting threads has been increased
438 * and we would wait on the SysV semaphore.
439 * A semaphore should wake up us more efficiently than
440 * a simple sched_yield() or usleep().
441 */
442
443 op.sem_num = 0;
444 op.sem_op = -1;
445 op.sem_flg = 0;
446
447 if (semop(m->semid, &op, 1) == -1) {
448 ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
449 "semop() failed while waiting on mutex %p", m);
450 ngx_abort();
451 }
452
453 ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
454 "mutex waked up %p lock:%XD", m, m->lock);
455
456 tries = 0;
457 old = m->lock;
458 continue;
459 }
460
461 old = m->lock;
462
463 } else {
464 lock = old | NGX_MUTEX_LOCK_BUSY;
465
466 if (ngx_atomic_cmp_set(&m->lock, old, lock)) {
467
468 /* we locked the mutex */
469
470 break;
471 }
472
473 old = m->lock;
474 }
475
476 if (tries++ > 1000) {
477
478 ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
479 "mutex %p is contested", m);
480
481 /* the mutex is probably contested so we are giving up now */
482
483 sched_yield();
484
485 tries = 0;
486 old = m->lock;
487 }
488 }
489
490 ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
491 "mutex %p is locked, lock:%XD", m, m->lock);
492
493 return NGX_OK;
494 }
495
496
497 void
498 ngx_mutex_unlock(ngx_mutex_t *m)
499 {
500 uint32_t lock, old;
501 struct sembuf op;
502
503 if (!ngx_threaded) {
504 return;
505 }
506
507 old = m->lock;
508
509 if (!(old & NGX_MUTEX_LOCK_BUSY)) {
510 ngx_log_error(NGX_LOG_ALERT, m->log, 0,
511 "trying to unlock the free mutex %p", m);
512 ngx_abort();
513 }
514
515 /* free the mutex */
516
517 #if 0
518 ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
519 "unlock mutex %p lock:%XD", m, old);
520 #endif
521
522 for ( ;; ) {
523 lock = old & ~NGX_MUTEX_LOCK_BUSY;
524
525 if (ngx_atomic_cmp_set(&m->lock, old, lock)) {
526 break;
527 }
528
529 old = m->lock;
530 }
531
532 if (m->semid == -1) {
533 ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
534 "mutex %p is unlocked", m);
535
536 return;
537 }
538
539 /* check whether we need to wake up a waiting thread */
540
541 old = m->lock;
542
543 for ( ;; ) {
544 if (old & NGX_MUTEX_LOCK_BUSY) {
545
546 /* the mutex is just locked by another thread */
547
548 break;
549 }
550
551 if (old == 0) {
552 break;
553 }
554
555 /* there are the waiting threads */
556
557 lock = old - 1;
558
559 if (ngx_atomic_cmp_set(&m->lock, old, lock)) {
560
561 /* wake up the thread that waits on semaphore */
562
563 ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
564 "wake up mutex %p", m);
565
566 op.sem_num = 0;
567 op.sem_op = 1;
568 op.sem_flg = 0;
569
570 if (semop(m->semid, &op, 1) == -1) {
571 ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
572 "semop() failed while waking up on mutex %p", m);
573 ngx_abort();
574 }
575
576 break;
577 }
578
579 old = m->lock;
580 }
581
582 ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
583 "mutex %p is unlocked", m);
584
585 return;
586 }
587
588
589 ngx_cond_t *
590 ngx_cond_init(ngx_log_t *log)
591 {
592 ngx_cond_t *cv;
593
594 cv = ngx_alloc(sizeof(ngx_cond_t), log);
595 if (cv == NULL) {
596 return NULL;
597 }
598
599 cv->signo = NGX_CV_SIGNAL;
600 cv->tid = -1;
601 cv->log = log;
602 cv->kq = -1;
603
604 return cv;
605 }
606
607
608 void
609 ngx_cond_destroy(ngx_cond_t *cv)
610 {
611 if (close(cv->kq) == -1) {
612 ngx_log_error(NGX_LOG_ALERT, cv->log, ngx_errno,
613 "kqueue close() failed");
614 }
615
616 ngx_free(cv);
617 }
618
619
620 ngx_int_t
621 ngx_cond_wait(ngx_cond_t *cv, ngx_mutex_t *m)
622 {
623 int n;
624 ngx_err_t err;
625 struct kevent kev;
626 struct timespec ts;
627
628 if (cv->kq == -1) {
629
630 /*
631 * We have to add the EVFILT_SIGNAL filter in the rfork()ed thread.
632 * Otherwise the thread would not get a signal event.
633 *
634 * However, we have not to open the kqueue in the thread,
635 * it is simply handy do it together.
636 */
637
638 cv->kq = kqueue();
639 if (cv->kq == -1) {
640 ngx_log_error(NGX_LOG_ALERT, cv->log, ngx_errno, "kqueue() failed");
641 return NGX_ERROR;
642 }
643
644 ngx_log_debug2(NGX_LOG_DEBUG_CORE, cv->log, 0,
645 "cv kq:%d signo:%d", cv->kq, cv->signo);
646
647 kev.ident = cv->signo;
648 kev.filter = EVFILT_SIGNAL;
649 kev.flags = EV_ADD;
650 kev.fflags = 0;
651 kev.data = 0;
652 kev.udata = NULL;
653
654 ts.tv_sec = 0;
655 ts.tv_nsec = 0;
656
657 if (kevent(cv->kq, &kev, 1, NULL, 0, &ts) == -1) {
658 ngx_log_error(NGX_LOG_ALERT, cv->log, ngx_errno, "kevent() failed");
659 return NGX_ERROR;
660 }
661
662 cv->tid = ngx_thread_self();
663 }
664
665 ngx_mutex_unlock(m);
666
667 ngx_log_debug3(NGX_LOG_DEBUG_CORE, cv->log, 0,
668 "cv %p wait, kq:%d, signo:%d", cv, cv->kq, cv->signo);
669
670 for ( ;; ) {
671 n = kevent(cv->kq, NULL, 0, &kev, 1, NULL);
672
673 ngx_log_debug2(NGX_LOG_DEBUG_CORE, cv->log, 0,
674 "cv %p kevent: %d", cv, n);
675
676 if (n == -1) {
677 err = ngx_errno;
678 ngx_log_error((err == NGX_EINTR) ? NGX_LOG_INFO : NGX_LOG_ALERT,
679 cv->log, ngx_errno,
680 "kevent() failed while waiting condition variable %p",
681 cv);
682
683 if (err == NGX_EINTR) {
684 break;
685 }
686
687 return NGX_ERROR;
688 }
689
690 if (n == 0) {
691 ngx_log_error(NGX_LOG_ALERT, cv->log, 0,
692 "kevent() returned no events "
693 "while waiting condition variable %p",
694 cv);
695 continue;
696 }
697
698 if (kev.filter != EVFILT_SIGNAL) {
699 ngx_log_error(NGX_LOG_ALERT, cv->log, 0,
700 "kevent() returned unexpected events: %d "
701 "while waiting condition variable %p",
702 kev.filter, cv);
703 continue;
704 }
705
706 if (kev.ident != (uintptr_t) cv->signo) {
707 ngx_log_error(NGX_LOG_ALERT, cv->log, 0,
708 "kevent() returned unexpected signal: %d ",
709 "while waiting condition variable %p",
710 kev.ident, cv);
711 continue;
712 }
713
714 break;
715 }
716
717 ngx_log_debug1(NGX_LOG_DEBUG_CORE, cv->log, 0, "cv %p is waked up", cv);
718
719 ngx_mutex_lock(m);
720
721 return NGX_OK;
722 }
723
724
725 ngx_int_t
726 ngx_cond_signal(ngx_cond_t *cv)
727 {
728 ngx_err_t err;
729
730 ngx_log_debug3(NGX_LOG_DEBUG_CORE, cv->log, 0,
731 "cv %p to signal %P %d",
732 cv, cv->tid, cv->signo);
733
734 if (cv->tid == -1) {
735 return NGX_OK;
736 }
737
738 if (kill(cv->tid, cv->signo) == -1) {
739
740 err = ngx_errno;
741
742 ngx_log_error(NGX_LOG_ALERT, cv->log, err,
743 "kill() failed while signaling condition variable %p", cv);
744
745 if (err == NGX_ESRCH) {
746 cv->tid = -1;
747 }
748
749 return NGX_ERROR;
750 }
751
752 ngx_log_debug1(NGX_LOG_DEBUG_CORE, cv->log, 0, "cv %p is signaled", cv);
753
754 return NGX_OK;
755 }
756
This page was automatically generated by the
LXR engine.
Visit the LXR main site for more
information.