Mercurial > hg > nginx
diff src/os/unix/ngx_freebsd_rfork_thread.c @ 266:5238e93961a1
nginx-0.0.2-2004-02-23-23:57:12 import
author | Igor Sysoev <igor@sysoev.ru> |
---|---|
date | Mon, 23 Feb 2004 20:57:12 +0000 |
parents | 0b67be7d4489 |
children | 83205e0b5522 |
line wrap: on
line diff
--- a/src/os/unix/ngx_freebsd_rfork_thread.c +++ b/src/os/unix/ngx_freebsd_rfork_thread.c @@ -2,31 +2,44 @@ #include <ngx_config.h> #include <ngx_core.h> - -extern int __isthreaded; +/* + * The threads implementation uses the rfork(RFPROC|RFTHREAD|RFMEM) + * to create threads. All threads use the stacks of the same size mmap()ed + * below the main stack. Thus the stack pointer is used to determine + * the current thread id. + * + * The mutex implementation uses the ngx_atomic_cmp_set() operation + * to acquire mutex and the SysV semaphore to wait on a mutex or to wake up + * the waiting threads. + * + * The condition variable implementation uses the SysV semaphore set of two + * semaphores. The first is used by the CV mutex, and the second is used + * by CV itself. + */ -typedef int ngx_tid_t; +extern int __isthreaded; static inline int ngx_gettid(); static char *usrstack; -static int red_zone = 4096; +static size_t rz_size = /* STUB: PAGE_SIZE */ 4096; static size_t stack_size; static size_t usable_stack_size; static char *last_stack; -static int threads; -static int nthreads; -static ngx_tid_t *tids; +static ngx_uint_t nthreads; +static ngx_uint_t max_threads; +static ngx_tid_t *tids; /* the threads tids array */ + /* the thread-safe errno */ static int errno0; /* the main thread's errno */ -static int *errnos; +static int *errnos; /* the threads errno's array */ int *__error() { @@ -41,20 +54,22 @@ int *__error() int ngx_create_thread(ngx_tid_t *tid, int (*func)(void *arg), void *arg, ngx_log_t *log) { - int id, err; - char *stack, *stack_top; + int id, err; + char *stack, *stack_top; - if (threads >= nthreads) { + if (nthreads >= max_threads) { ngx_log_error(NGX_LOG_CRIT, log, 0, - "no more than %d threads can be created", nthreads); + "no more than %d threads can be created", max_threads); return NGX_ERROR; } last_stack -= stack_size; + stack = mmap(last_stack, usable_stack_size, PROT_READ|PROT_WRITE, MAP_STACK, -1, 0); + if (stack == MAP_FAILED) { - ngx_log_error(NGX_LOG_ALERT, log, errno, + ngx_log_error(NGX_LOG_ALERT, log, ngx_errno, "mmap(%08X:%d, MAP_STACK) thread stack failed", last_stack, usable_stack_size); return NGX_ERROR; @@ -66,7 +81,8 @@ int ngx_create_thread(ngx_tid_t *tid, in stack_top = stack + usable_stack_size; -printf("stack: %08X-%08X\n", stack, stack_top); + ngx_log_debug2(NGX_LOG_DEBUG_CORE, log, 0, + "thread stack: %08X-%08X", stack, stack_top); #if 1 id = rfork_thread(RFPROC|RFTHREAD|RFMEM, stack_top, func, arg); @@ -78,92 +94,81 @@ printf("stack: %08X-%08X\n", stack, stac id = rfork(RFFDG|RFCFDG); #endif - err = errno; + err = ngx_errno; if (id == -1) { ngx_log_error(NGX_LOG_ALERT, log, err, "rfork() failed"); } else { *tid = id; - threads = (usrstack - stack_top) / stack_size; - tids[threads] = id; + nthreads = (usrstack - stack_top) / stack_size; + tids[nthreads] = id; - /* allow the spinlock in libc malloc() */ - __isthreaded = 1; + ngx_log_debug1(NGX_LOG_DEBUG_CORE, log, 0, "rfork()ed thread: %d", id); } return err; } -int ngx_init_thread_env(int n, size_t size, ngx_log_t *log) +ngx_int_t ngx_init_threads(int n, size_t size, ngx_log_t *log) { int len; - char *rz, *zone; + char *red_zone, *zone; - nthreads = n; + max_threads = n; len = 4; if (sysctlbyname("kern.usrstack", &usrstack, &len, NULL, 0) == -1) { - ngx_log_error(NGX_LOG_ALERT, log, errno, + ngx_log_error(NGX_LOG_ALERT, log, ngx_errno, "sysctlbyname(kern.usrstack) failed"); return NGX_ERROR; } -printf("usrstack: %08X\n", usrstack); + /* the main thread stack red zone */ + red_zone = usrstack - (size + rz_size); - /* red zone */ - rz = usrstack - (size + red_zone); + ngx_log_debug2(NGX_LOG_DEBUG_CORE, log, 0, + "usrstack: %08X, red zone: %08X", usrstack, red_zone); -printf("red zone: %08X\n", rz); - - zone = mmap(rz, red_zone, PROT_NONE, MAP_ANON, -1, 0); + zone = mmap(red_zone, rz_size, PROT_NONE, MAP_ANON, -1, 0); if (zone == MAP_FAILED) { - ngx_log_error(NGX_LOG_ALERT, log, errno, + ngx_log_error(NGX_LOG_ALERT, log, ngx_errno, "mmap(%08X:%d, PROT_NONE, MAP_ANON) red zone failed", - rz, red_zone); + red_zone, rz_size); return NGX_ERROR; } - if (zone != rz) { + if (zone != red_zone) { ngx_log_error(NGX_LOG_ALERT, log, 0, "red zone address was changed"); } - /* create the thread errno array */ - ngx_test_null(errnos, ngx_calloc(n * sizeof(int), log), NGX_ERROR); + /* create the threads errno array */ - /* create the thread tid array */ - ngx_test_null(tids, ngx_calloc((n + 1) * sizeof(ngx_tid_t), log), - NGX_ERROR); + if (!(errnos = ngx_calloc(n * sizeof(int), log))) { + return NGX_ERROR; + } + + /* create the threads tid array */ - tids[0] = ngx_getpid(); - threads = 1; + if (!(tids = ngx_calloc((n + 1) * sizeof(ngx_tid_t), log))) { + return NGX_ERROR; + } + + tids[0] = ngx_pid; + nthreads = 1; - last_stack = zone + red_zone; + last_stack = zone + rz_size; usable_stack_size = size; - stack_size = size + red_zone; + stack_size = size + rz_size; + + /* allow the spinlock in libc malloc() */ + __isthreaded = 1; return NGX_OK; } -ngx_tid_t ngx_thread_self() -{ - int tid; - ngx_tid_t pid; - - tid = ngx_gettid(); - - if (tids[tid] == 0) { - pid = ngx_getpid(); - tids[tid] = pid; - return pid; - } - - return tids[tid]; -} - - static inline int ngx_gettid() { char *sp; @@ -176,3 +181,286 @@ static inline int ngx_gettid() return (usrstack - sp) / stack_size; } + + +ngx_tid_t ngx_thread_self() +{ + int tid; + ngx_tid_t pid; + + tid = ngx_gettid(); + + if (tids == NULL) { + return ngx_pid; + } + +#if 0 + if (tids[tid] == 0) { + pid = ngx_pid; + tids[tid] = pid; + return pid; + } +#endif + + return tids[tid]; +} + + +ngx_mutex_t *ngx_mutex_init(ngx_log_t *log, uint flags) +{ + int nsem, i; + ngx_mutex_t *m; + union semun op; + + if (!(m = ngx_alloc(sizeof(ngx_mutex_t), log))) { + return NULL; + } + + m->lock = 0; + m->log = log; + + if (flags & NGX_MUTEX_LIGHT) { + m->semid = -1; + return m; + } + + nsem = flags & NGX_MUTEX_CV ? 2 : 1; + + m->semid = semget(IPC_PRIVATE, nsem, SEM_R|SEM_A); + if (m->semid == -1) { + ngx_log_error(NGX_LOG_ALERT, log, ngx_errno, "semget() failed"); + return NULL; + } + + op.val = 0; + for (i = 0; i < nsem; i++) { + if (semctl(m->semid, i, SETVAL, op) == -1) { + ngx_log_error(NGX_LOG_ALERT, log, ngx_errno, + "semctl(SETVAL) failed"); + + if (semctl(m->semid, 0, IPC_RMID) == -1) { + ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno, + "semctl(IPC_RMID) failed"); + } + + return NULL; + } + } + + return m; +} + + +void ngx_mutex_done(ngx_mutex_t *m) +{ + if (semctl(m->semid, 0, IPC_RMID) == -1) { + ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno, + "semctl(IPC_RMID) failed"); + } + + ngx_free(m); +} + + +ngx_int_t ngx_mutex_do_lock(ngx_mutex_t *m, ngx_int_t try) +{ + uint32_t lock, new, old; + ngx_uint_t tries; + struct sembuf op; + +#if (NGX_DEBUG) + if (try) { + ngx_log_debug2(NGX_LOG_DEBUG_CORE, m->log, 0, + "try lock mutex %08X lock:%X", m, m->lock); + } else { + ngx_log_debug2(NGX_LOG_DEBUG_CORE, m->log, 0, + "lock mutex %08X lock:%X", m, m->lock); + } +#endif + + old = m->lock; + tries = 0; + + for ( ;; ) { + if (old & NGX_MUTEX_LOCK_BUSY) { + + if (try) { + return NGX_AGAIN; + } + + if (ngx_freebsd_hw_ncpu > 1 && tries++ < 1000) { + + /* the spinlock is used only on the SMP system */ + + old = m->lock; + continue; + } + + if (m->semid == -1) { + sched_yield(); + + tries = 0; + old = m->lock; + continue; + } + + ngx_log_debug2(NGX_LOG_DEBUG_CORE, m->log, 0, + "mutex %08X lock:%X", m, m->lock); + + /* + * The mutex is locked so we increase a number + * of the threads that are waiting on the mutex + */ + + lock = old + 1; + + if ((lock & ~NGX_MUTEX_LOCK_BUSY) > nthreads) { + ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno, + "%d threads wait for mutex %0X, " + "while only %d threads are available", + lock & ~NGX_MUTEX_LOCK_BUSY, m, nthreads); + return NGX_ERROR; + } + + if (ngx_atomic_cmp_set(&m->lock, old, lock)) { + + ngx_log_debug2(NGX_LOG_DEBUG_CORE, m->log, 0, + "wait mutex %08X lock:%X", m, m->lock); + + /* + * The number of the waiting threads has been increased + * and we would wait on the SysV semaphore. + * A semaphore should wake up us more efficiently than + * a simple usleep(). + */ + + op.sem_num = 0; + op.sem_op = -1; + op.sem_flg = SEM_UNDO; + + if (semop(m->semid, &op, 1) == -1) { + ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno, + "semop() failed while waiting " + "on mutex %08X", m); + return NGX_ERROR; + } + + tries = 0; + old = m->lock; + continue; + } + + old = m->lock; + + } else { + lock = old | NGX_MUTEX_LOCK_BUSY; + + if (ngx_atomic_cmp_set(&m->lock, old, lock)) { + + /* we locked the mutex */ + + break; + } + + old = m->lock; + } + + if (tries++ > 1000) { + + ngx_log_debug1(NGX_LOG_DEBUG_CORE, m->log, 0, + "mutex %08X is contested", m); + + /* the mutex is probably contested so we are giving up now */ + + sched_yield(); + + tries = 0; + old = m->lock; + } + } + + ngx_log_debug2(NGX_LOG_DEBUG_CORE, m->log, 0, + "mutex %08X is locked, lock:%X", m, m->lock); + + return NGX_OK; +} + + +ngx_int_t ngx_mutex_unlock(ngx_mutex_t *m) +{ + uint32_t lock, new, old; + struct sembuf op; + + old = m->lock; + + if (!(old & NGX_MUTEX_LOCK_BUSY)) { + ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno, + "tring to unlock the free mutex %0X", m); + return NGX_ERROR; + } + + /* free the mutex */ + + for ( ;; ) { + lock = old & ~NGX_MUTEX_LOCK_BUSY; + + if (ngx_atomic_cmp_set(&m->lock, old, lock)) { + break; + } + + old = m->lock; + } + + if (m->semid == -1) { + ngx_log_debug1(NGX_LOG_DEBUG_CORE, m->log, 0, + "mutex %08X is unlocked", m); + + return NGX_OK; + } + + /* check weather we need to wake up a waiting thread */ + + old = m->lock; + + for ( ;; ) { + if (old & NGX_MUTEX_LOCK_BUSY) { + + /* the mutex is just locked by another thread */ + + break; + } + + if (old == 0) { + break; + } + + /* there are the waiting threads */ + + lock = old - 1; + + if (ngx_atomic_cmp_set(&m->lock, old, lock)) { + + /* wake up the thread that waits on semaphore */ + + op.sem_num = 0; + op.sem_op = 1; + op.sem_flg = SEM_UNDO; + + if (semop(m->semid, &op, 1) == -1) { + ngx_log_error(NGX_LOG_ALERT, m->log, ngx_errno, + "semop() failed while waking up on mutex %08X", + m); + return NGX_ERROR; + } + + break; + } + + old = m->lock; + } + + ngx_log_debug1(NGX_LOG_DEBUG_CORE, m->log, 0, + "mutex %08X is unlocked", m); + + return NGX_OK; +}