One Level Up Top Level

src/os/unix/ngx_freebsd_rfork_thread.c - nginx-1.7.10

Global variables defined

Functions defined

Source code


  2. /*
  3. * Copyright (C) Igor Sysoev
  4. * Copyright (C) Nginx, Inc.
  5. */


  8. #include <ngx_config.h>
  9. #include <ngx_core.h>

  11. /*
  12. * The threads implementation uses the rfork(RFPROC|RFTHREAD|RFMEM) syscall
  13. * to create threads.  All threads use the stacks of the same size mmap()ed
  14. * below the main stack.  Thus the current thread id is determined via
  15. * the stack pointer value.
  16. *
  17. * The mutex implementation uses the ngx_atomic_cmp_set() operation
  18. * to acquire a mutex and the SysV semaphore to wait on a mutex and to wake up
  19. * the waiting threads.  The light mutex does not use semaphore, so after
  20. * spinning in the lock the thread calls sched_yield().  However the light
  21. * mutexes are intended to be used with the "trylock" operation only.
  22. * The SysV semop() is a cheap syscall, particularly if it has little sembuf's
  23. * and does not use SEM_UNDO.
  24. *
  25. * The condition variable implementation uses the signal #64.
  26. * The signal handler is SIG_IGN so the kill() is a cheap syscall.
  27. * The thread waits a signal in kevent().  The use of the EVFILT_SIGNAL
  28. * is safe since FreeBSD 4.10-STABLE.
  29. *
  30. * This threads implementation currently works on i386 (486+) and amd64
  31. * platforms only.
  32. */


  35. ‌char                 *ngx_freebsd_kern_usrstack;
  36. ‌size_t                ngx_thread_stack_size;


  39. ‌static size_t         rz_size;
  40. ‌static size_t         usable_stack_size;
  41. ‌static char          *last_stack;

  43. ‌static ngx_uint_t     nthreads;
  44. ‌static ngx_uint_t     max_threads;

  46. ‌static ngx_uint_t     nkeys;
  47. ‌static ngx_tid_t     *tids;      /* the threads tids array */
  48. ‌void                **ngx_tls;   /* the threads tls's array */

  50. /* the thread-safe libc errno */

  52. ‌static int   errno0;   /* the main thread's errno */
  53. ‌static int  *errnos;   /* the threads errno's array */

  55. int *
  56. ‌__error()
  57. {
  58.     int  tid;

  60.     tid = ngx_gettid();

  62.     return tid ? &errnos[tid - 1] : &errno0;
  63. }


  66. /*
  67. * __isthreaded enables the spinlocks in some libc functions, i.e. in malloc()
  68. * and some other places.  Nevertheless we protect our malloc()/free() calls
  69. * by own mutex that is more efficient than the spinlock.
  70. *
  71. * _spinlock() is a weak referenced stub in src/lib/libc/gen/_spinlock_stub.c
  72. * that does nothing.
  73. */

  75. extern int  __isthreaded;

  77. void
  78. ‌_spinlock(ngx_atomic_t *lock)
  79. {
  80.     ngx_int_t  tries;

  82.     tries = 0;

  84.     for ( ;; ) {

  86.         if (*lock) {
  87.             if (ngx_ncpu > 1 && tries++ < 1000) {
  88.                 continue;
  89.             }

  91.             sched_yield();
  92.             tries = 0;

  94.         } else {
  95.             if (ngx_atomic_cmp_set(lock, 0, 1)) {
  96.                 return;
  97.             }
  98.         }
  99.     }
  100. }


  103. /*
  104. * Before FreeBSD 5.1 _spinunlock() is a simple #define in
  105. * src/lib/libc/include/spinlock.h that zeroes lock.
  106. *
  107. * Since FreeBSD 5.1 _spinunlock() is a weak referenced stub in
  108. * src/lib/libc/gen/_spinlock_stub.c that does nothing.
  109. */

  111. #ifndef _spinunlock

  113. void
  114. ‌_spinunlock(ngx_atomic_t *lock)
  115. {
  116.     *lock = 0;
  117. }

  119. #endif


  122. ngx_err_t
  123. ‌ngx_create_thread(ngx_tid_t *tid, ngx_thread_value_t (*func)(void *arg),
  124.     void *arg, ngx_log_t *log)
  125. {
  126.     ngx_pid_t   id;
  127.     ngx_err_t   err;
  128.     char       *stack, *stack_top;

  130.     if (nthreads >= max_threads) {
  131.         ngx_log_error(NGX_LOG_CRIT, log, 0,
  132.                       "no more than %ui threads can be created", max_threads);
  133.         return NGX_ERROR;
  134.     }

  136.     last_stack -= ngx_thread_stack_size;

  138.     stack = mmap(last_stack, usable_stack_size, PROT_READ|PROT_WRITE,
  139.                  MAP_STACK, -1, 0);

  141.     if (stack == MAP_FAILED) {
  142.         ngx_log_error(NGX_LOG_ALERT, log, ngx_errno,
  143.                       "mmap(%p:%uz, MAP_STACK) thread stack failed",
  144.                       last_stack, usable_stack_size);
  145.         return NGX_ERROR;
  146.     }

  148.     if (stack != last_stack) {
  149.         ngx_log_error(NGX_LOG_ALERT, log, 0,
  150.                       "stack %p address was changed to %p", last_stack, stack);
  151.         return NGX_ERROR;
  152.     }

  154.     stack_top = stack + usable_stack_size;

  156.     ngx_log_debug2(NGX_LOG_DEBUG_CORE, log, 0,
  157.                    "thread stack: %p-%p", stack, stack_top);

  159.     ngx_set_errno(0);

  161.     id = rfork_thread(RFPROC|RFTHREAD|RFMEM, stack_top,
  162.                       (ngx_rfork_thread_func_pt) func, arg);

  164.     err = ngx_errno;

  166.     if (id == -1) {
  167.         ngx_log_error(NGX_LOG_ALERT, log, err, "rfork() failed");

  169.     } else {
  170.         *tid = id;
  171.         nthreads = (ngx_freebsd_kern_usrstack - stack_top)
  172.                                                        / ngx_thread_stack_size;
  173.         tids[nthreads] = id;

  175.         ngx_log_debug1(NGX_LOG_DEBUG_CORE, log, 0, "rfork()ed thread: %P", id);
  176.     }

  178.     return err;
  179. }


  182. ngx_int_t
  183. ‌ngx_init_threads(int n, size_t size, ngx_cycle_t *cycle)
  184. {
  185.     char              *red_zone, *zone;
  186.     size_t             len;
  187.     ngx_int_t          i;
  188.     struct sigaction   sa;

  190.     max_threads = n + 1;

  192.     for (i = 0; i < n; i++) {
  193.         ngx_memzero(&sa, sizeof(struct sigaction));
  194.         sa.sa_handler = SIG_IGN;
  195.         sigemptyset(&sa.sa_mask);
  196.         if (sigaction(NGX_CV_SIGNAL, &sa, NULL) == -1) {
  197.             ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
  198.                           "sigaction(%d, SIG_IGN) failed", NGX_CV_SIGNAL);
  199.             return NGX_ERROR;
  200.         }
  201.     }

  203.     len = sizeof(ngx_freebsd_kern_usrstack);
  204.     if (sysctlbyname("kern.usrstack", &ngx_freebsd_kern_usrstack, &len,
  205.                                                                 NULL, 0) == -1)
  206.     {
  207.         ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
  208.                       "sysctlbyname(kern.usrstack) failed");
  209.         return NGX_ERROR;
  210.     }

  212.     /* the main thread stack red zone */
  213.     rz_size = ngx_pagesize;
  214.     red_zone = ngx_freebsd_kern_usrstack - (size + rz_size);

  216.     ngx_log_debug2(NGX_LOG_DEBUG_CORE, cycle->log, 0,
  217.                    "usrstack: %p red zone: %p",
  218.                    ngx_freebsd_kern_usrstack, red_zone);

  220.     zone = mmap(red_zone, rz_size, PROT_NONE, MAP_ANON, -1, 0);
  221.     if (zone == MAP_FAILED) {
  222.         ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno,
  223.                       "mmap(%p:%uz, PROT_NONE, MAP_ANON) red zone failed",
  224.                       red_zone, rz_size);
  225.         return NGX_ERROR;
  226.     }

  228.     if (zone != red_zone) {
  229.         ngx_log_error(NGX_LOG_ALERT, cycle->log, 0,
  230.                       "red zone %p address was changed to %p", red_zone, zone);
  231.         return NGX_ERROR;
  232.     }

  234.     /* create the thread errno' array */

  236.     errnos = ngx_calloc(n * sizeof(int), cycle->log);
  237.     if (errnos == NULL) {
  238.         return NGX_ERROR;
  239.     }

  241.     /* create the thread tids array */

  243.     tids = ngx_calloc((n + 1) * sizeof(ngx_tid_t), cycle->log);
  244.     if (tids == NULL) {
  245.         return NGX_ERROR;
  246.     }

  248.     tids[0] = ngx_pid;

  250.     /* create the thread tls' array */

  252.     ngx_tls = ngx_calloc(NGX_THREAD_KEYS_MAX * (n + 1) * sizeof(void *),
  253.                          cycle->log);
  254.     if (ngx_tls == NULL) {
  255.         return NGX_ERROR;
  256.     }

  258.     nthreads = 1;

  260.     last_stack = zone + rz_size;
  261.     usable_stack_size = size;
  262.     ngx_thread_stack_size = size + rz_size;

  264.     /* allow the spinlock in libc malloc() */
  265.     __isthreaded = 1;

  267.     ngx_threaded = 1;

  269.     return NGX_OK;
  270. }


  273. ngx_tid_t
  274. ‌ngx_thread_self(void)
  275. {
  276.     ngx_int_t  tid;

  278.     tid = ngx_gettid();

  280.     if (tids == NULL) {
  281.         return ngx_pid;
  282.     }

  284.     return tids[tid];
  285. }


  288. ngx_err_t
  289. ‌ngx_thread_key_create(ngx_tls_key_t *key)
  290. {
  291.     if (nkeys >= NGX_THREAD_KEYS_MAX) {
  292.         return NGX_ENOMEM;
  293.     }

  295.     *key = nkeys++;

  297.     return 0;
  298. }


  301. ngx_err_t
  302. ‌ngx_thread_set_tls(ngx_tls_key_t key, void *value)
  303. {
  304.     if (key >= NGX_THREAD_KEYS_MAX) {
  305.         return NGX_EINVAL;
  306.     }

  308.     ngx_tls[key * NGX_THREAD_KEYS_MAX + ngx_gettid()] = value;
  309.     return 0;
  310. }


  313. ngx_mutex_t *
  314. ‌ngx_mutex_init(ngx_log_t *log, ngx_uint_t flags)
  315. {
  316.     ngx_mutex_t  *m;
  317.     union semun   op;

  319.     m = ngx_alloc(sizeof(ngx_mutex_t), log);
  320.     if (m == NULL) {
  321.         return NULL;
  322.     }

  324.     m->lock = 0;
  325.     m->log = log;

  327.     if (flags & NGX_MUTEX_LIGHT) {
  328.         m->semid = -1;
  329.         return m;
  330.     }

  332.     m->semid = semget(IPC_PRIVATE, 1, SEM_R|SEM_A);
  333.     if (m->semid == -1) {
  334.         ngx_log_error(NGX_LOG_ALERT, log, ngx_errno, "semget() failed");
  335.         return NULL;
  336.     }

  338.     op.val = 0;

  340.     if (semctl(m->semid, 0, SETVAL, op) == -1) {
  341.         ngx_log_error(NGX_LOG_ALERT, log, ngx_errno, "semctl(SETVAL) failed");

  343.         if (semctl(m->semid, 0, IPC_RMID) == -1) {
  344.             ngx_log_error(NGX_LOG_ALERT, log, ngx_errno,
  345.                           "semctl(IPC_RMID) failed");
  346.         }

  348.         return NULL;
  349.     }

  351.     return m;
  352. }


  355. void
  356. ‌ngx_mutex_destroy(ngx_mutex_t *m)
  357. {
  358.     if (semctl(m->semid, 0, IPC_RMID) == -1) {
  359.         ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
  360.                       "semctl(IPC_RMID) failed");
  361.     }

  363.     ngx_free((void *) m);
  364. }


  367. ngx_int_t
  368. ‌ngx_mutex_dolock(ngx_mutex_t *m, ngx_int_t try)
  369. {
  370.     uint32_t       lock, old;
  371.     ngx_uint_t     tries;
  372.     struct sembuf  op;

  374.     if (!ngx_threaded) {
  375.         return NGX_OK;
  376.     }

  378. #if (NGX_DEBUG)
  379.     if (try) {
  380.         ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  381.                        "try lock mutex %p lock:%XD", m, m->lock);
  382.     } else {
  383.         ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  384.                        "lock mutex %p lock:%XD", m, m->lock);
  385.     }
  386. #endif

  388.     old = m->lock;
  389.     tries = 0;

  391.     for ( ;; ) {
  392.         if (old & NGX_MUTEX_LOCK_BUSY) {

  394.             if (try) {
  395.                 return NGX_AGAIN;
  396.             }

  398.             if (ngx_ncpu > 1 && tries++ < 1000) {

  400.                 /* the spinlock is used only on the SMP system */

  402.                 old = m->lock;
  403.                 continue;
  404.             }

  406.             if (m->semid == -1) {
  407.                 sched_yield();

  409.                 tries = 0;
  410.                 old = m->lock;
  411.                 continue;
  412.             }

  414.             ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  415.                            "mutex %p lock:%XD", m, m->lock);

  417.             /*
  418.              * The mutex is locked so we increase a number
  419.              * of the threads that are waiting on the mutex
  420.              */

  422.             lock = old + 1;

  424.             if ((lock & ~NGX_MUTEX_LOCK_BUSY) > nthreads) {
  425.                 ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
  426.                               "%D threads wait for mutex %p, "
  427.                               "while only %ui threads are available",
  428.                               lock & ~NGX_MUTEX_LOCK_BUSY, m, nthreads);
  429.                 ngx_abort();
  430.             }

  432.             if (ngx_atomic_cmp_set(&m->lock, old, lock)) {

  434.                 ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  435.                                "wait mutex %p lock:%XD", m, m->lock);

  437.                 /*
  438.                  * The number of the waiting threads has been increased
  439.                  * and we would wait on the SysV semaphore.
  440.                  * A semaphore should wake up us more efficiently than
  441.                  * a simple sched_yield() or usleep().
  442.                  */

  444.                 op.sem_num = 0;
  445.                 op.sem_op = -1;
  446.                 op.sem_flg = 0;

  448.                 if (semop(m->semid, &op, 1) == -1) {
  449.                     ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
  450.                                  "semop() failed while waiting on mutex %p", m);
  451.                     ngx_abort();
  452.                 }

  454.                 ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  455.                                "mutex waked up %p lock:%XD", m, m->lock);

  457.                 tries = 0;
  458.                 old = m->lock;
  459.                 continue;
  460.             }

  462.             old = m->lock;

  464.         } else {
  465.             lock = old | NGX_MUTEX_LOCK_BUSY;

  467.             if (ngx_atomic_cmp_set(&m->lock, old, lock)) {

  469.                 /* we locked the mutex */

  471.                 break;
  472.             }

  474.             old = m->lock;
  475.         }

  477.         if (tries++ > 1000) {

  479.             ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  480.                            "mutex %p is contested", m);

  482.             /* the mutex is probably contested so we are giving up now */

  484.             sched_yield();

  486.             tries = 0;
  487.             old = m->lock;
  488.         }
  489.     }

  491.     ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  492.                    "mutex %p is locked, lock:%XD", m, m->lock);

  494.     return NGX_OK;
  495. }


  498. void
  499. ‌ngx_mutex_unlock(ngx_mutex_t *m)
  500. {
  501.     uint32_t       lock, old;
  502.     struct sembuf  op;

  504.     if (!ngx_threaded) {
  505.         return;
  506.     }

  508.     old = m->lock;

  510.     if (!(old & NGX_MUTEX_LOCK_BUSY)) {
  511.         ngx_log_error(NGX_LOG_ALERT, m->log, 0,
  512.                       "trying to unlock the free mutex %p", m);
  513.         ngx_abort();
  514.     }

  516.     /* free the mutex */

  518. #if 0
  519.     ngx_log_debug2(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  520.                    "unlock mutex %p lock:%XD", m, old);
  521. #endif

  523.     for ( ;; ) {
  524.         lock = old & ~NGX_MUTEX_LOCK_BUSY;

  526.         if (ngx_atomic_cmp_set(&m->lock, old, lock)) {
  527.             break;
  528.         }

  530.         old = m->lock;
  531.     }

  533.     if (m->semid == -1) {
  534.         ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  535.                        "mutex %p is unlocked", m);

  537.         return;
  538.     }

  540.     /* check whether we need to wake up a waiting thread */

  542.     old = m->lock;

  544.     for ( ;; ) {
  545.         if (old & NGX_MUTEX_LOCK_BUSY) {

  547.             /* the mutex is just locked by another thread */

  549.             break;
  550.         }

  552.         if (old == 0) {
  553.             break;
  554.         }

  556.         /* there are the waiting threads */

  558.         lock = old - 1;

  560.         if (ngx_atomic_cmp_set(&m->lock, old, lock)) {

  562.             /* wake up the thread that waits on semaphore */

  564.             ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  565.                            "wake up mutex %p", m);

  567.             op.sem_num = 0;
  568.             op.sem_op = 1;
  569.             op.sem_flg = 0;

  571.             if (semop(m->semid, &op, 1) == -1) {
  572.                 ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno,
  573.                               "semop() failed while waking up on mutex %p", m);
  574.                 ngx_abort();
  575.             }

  577.             break;
  578.         }

  580.         old = m->lock;
  581.     }

  583.     ngx_log_debug1(NGX_LOG_DEBUG_MUTEX, m->log, 0,
  584.                    "mutex %p is unlocked", m);

  586.     return;
  587. }


  590. ngx_cond_t *
  591. ‌ngx_cond_init(ngx_log_t *log)
  592. {
  593.     ngx_cond_t  *cv;

  595.     cv = ngx_alloc(sizeof(ngx_cond_t), log);
  596.     if (cv == NULL) {
  597.         return NULL;
  598.     }

  600.     cv->signo = NGX_CV_SIGNAL;
  601.     cv->tid = -1;
  602.     cv->log = log;
  603.     cv->kq = -1;

  605.     return cv;
  606. }


  609. void
  610. ‌ngx_cond_destroy(ngx_cond_t *cv)
  611. {
  612.     if (close(cv->kq) == -1) {
  613.         ngx_log_error(NGX_LOG_ALERT, cv->log, ngx_errno,
  614.                       "kqueue close() failed");
  615.     }

  617.     ngx_free(cv);
  618. }


  621. ngx_int_t
  622. ‌ngx_cond_wait(ngx_cond_t *cv, ngx_mutex_t *m)
  623. {
  624.     int              n;
  625.     ngx_err_t        err;
  626.     struct kevent    kev;
  627.     struct timespec  ts;

  629.     if (cv->kq == -1) {

  631.         /*
  632.          * We have to add the EVFILT_SIGNAL filter in the rfork()ed thread.
  633.          * Otherwise the thread would not get a signal event.
  634.          *
  635.          * However, we have not to open the kqueue in the thread,
  636.          * it is simply handy do it together.
  637.          */

  639.         cv->kq = kqueue();
  640.         if (cv->kq == -1) {
  641.             ngx_log_error(NGX_LOG_ALERT, cv->log, ngx_errno, "kqueue() failed");
  642.             return NGX_ERROR;
  643.         }

  645.         ngx_log_debug2(NGX_LOG_DEBUG_CORE, cv->log, 0,
  646.                        "cv kq:%d signo:%d", cv->kq, cv->signo);

  648.         kev.ident = cv->signo;
  649.         kev.filter = EVFILT_SIGNAL;
  650.         kev.flags = EV_ADD;
  651.         kev.fflags = 0;
  652.         kev.data = 0;
  653.         kev.udata = NULL;

  655.         ts.tv_sec = 0;
  656.         ts.tv_nsec = 0;

  658.         if (kevent(cv->kq, &kev, 1, NULL, 0, &ts) == -1) {
  659.             ngx_log_error(NGX_LOG_ALERT, cv->log, ngx_errno, "kevent() failed");
  660.             return NGX_ERROR;
  661.         }

  663.         cv->tid = ngx_thread_self();
  664.     }

  666.     ngx_mutex_unlock(m);

  668.     ngx_log_debug3(NGX_LOG_DEBUG_CORE, cv->log, 0,
  669.                    "cv %p wait, kq:%d, signo:%d", cv, cv->kq, cv->signo);

  671.     for ( ;; ) {
  672.         n = kevent(cv->kq, NULL, 0, &kev, 1, NULL);

  674.         ngx_log_debug2(NGX_LOG_DEBUG_CORE, cv->log, 0,
  675.                        "cv %p kevent: %d", cv, n);

  677.         if (n == -1) {
  678.             err = ngx_errno;
  679.             ngx_log_error((err == NGX_EINTR) ? NGX_LOG_INFO : NGX_LOG_ALERT,
  680.                           cv->log, ngx_errno,
  681.                           "kevent() failed while waiting condition variable %p",
  682.                           cv);

  684.             if (err == NGX_EINTR) {
  685.                 break;
  686.             }

  688.             return NGX_ERROR;
  689.         }

  691.         if (n == 0) {
  692.             ngx_log_error(NGX_LOG_ALERT, cv->log, 0,
  693.                           "kevent() returned no events "
  694.                           "while waiting condition variable %p",
  695.                           cv);
  696.             continue;
  697.         }

  699.         if (kev.filter != EVFILT_SIGNAL) {
  700.             ngx_log_error(NGX_LOG_ALERT, cv->log, 0,
  701.                           "kevent() returned unexpected events: %d "
  702.                           "while waiting condition variable %p",
  703.                           kev.filter, cv);
  704.             continue;
  705.         }

  707.         if (kev.ident != (uintptr_t) cv->signo) {
  708.             ngx_log_error(NGX_LOG_ALERT, cv->log, 0,
  709.                           "kevent() returned unexpected signal: %d ",
  710.                           "while waiting condition variable %p",
  711.                           kev.ident, cv);
  712.             continue;
  713.         }

  715.         break;
  716.     }

  718.     ngx_log_debug1(NGX_LOG_DEBUG_CORE, cv->log, 0, "cv %p is waked up", cv);

  720.     ngx_mutex_lock(m);

  722.     return NGX_OK;
  723. }


  726. ngx_int_t
  727. ‌ngx_cond_signal(ngx_cond_t *cv)
  728. {
  729.     ngx_err_t  err;

  731.     ngx_log_debug3(NGX_LOG_DEBUG_CORE, cv->log, 0,
  732.                    "cv %p to signal %P %d",
  733.                    cv, cv->tid, cv->signo);

  735.     if (cv->tid == -1) {
  736.         return NGX_OK;
  737.     }

  739.     if (kill(cv->tid, cv->signo) == -1) {

  741.         err = ngx_errno;

  743.         ngx_log_error(NGX_LOG_ALERT, cv->log, err,
  744.                      "kill() failed while signaling condition variable %p", cv);

  746.         if (err == NGX_ESRCH) {
  747.             cv->tid = -1;
  748.         }

  750.         return NGX_ERROR;
  751.     }

  753.     ngx_log_debug1(NGX_LOG_DEBUG_CORE, cv->log, 0, "cv %p is signaled", cv);

  755.     return NGX_OK;
  756. }

One Level Up Top Level