static void ngx_master_process_cycle(ngx_cycle_t *cycle, ngx_master_ctx_t *ctx);
static void ngx_master_exit(ngx_cycle_t *cycle, ngx_master_ctx_t *ctx);
static void ngx_worker_process_cycle(ngx_cycle_t *cycle, void *data);
+#if (NGX_THREADS)
+static int ngx_worker_thread_cycle(void *data);
+#endif
static ngx_int_t ngx_add_inherited_sockets(ngx_cycle_t *cycle, char **envp);
static ngx_pid_t ngx_exec_new_binary(ngx_cycle_t *cycle, char *const *argv);
static ngx_int_t ngx_getopt(ngx_master_ctx_t *ctx, ngx_cycle_t *cycle);
};
-ngx_int_t ngx_max_module;
-ngx_uint_t ngx_connection_counter;
+ngx_int_t ngx_max_module;
+ngx_atomic_t ngx_connection_counter;
-ngx_int_t ngx_process;
-ngx_pid_t ngx_pid;
-ngx_pid_t ngx_new_binary;
+ngx_int_t ngx_process;
+ngx_pid_t ngx_pid;
+ngx_pid_t ngx_new_binary;
-ngx_int_t ngx_inherited;
-ngx_int_t ngx_reap;
-ngx_int_t ngx_timer;
-ngx_int_t ngx_terminate;
-ngx_int_t ngx_quit;
-ngx_int_t ngx_noaccept;
-ngx_int_t ngx_reconfigure;
-ngx_int_t ngx_reopen;
-ngx_int_t ngx_change_binary;
+ngx_int_t ngx_inherited;
+ngx_int_t ngx_reap;
+ngx_int_t ngx_timer;
+ngx_int_t ngx_terminate;
+ngx_int_t ngx_quit;
+ngx_int_t ngx_noaccept;
+ngx_int_t ngx_reconfigure;
+ngx_int_t ngx_reopen;
+ngx_int_t ngx_change_binary;
int main(int argc, char *const *argv, char **envp)
ngx_int_t i;
ngx_listening_t *ls;
ngx_core_conf_t *ccf;
+#if (NGX_THREADS)
+ ngx_tid_t tid;
+#endif
ngx_process = NGX_PROCESS_WORKER;
ngx_last_process = 0;
ngx_setproctitle("worker process");
- /* TODO: threads: start ngx_worker_thread_cycle() */
+#if (NGX_THREADS)
+
+ ngx_init_threads(5, 128 * 1024 * 1024, cycle->log);
+
+ for (i = 0; i < 1; i++) {
+ ngx_create_thread(&tid, ngx_worker_thread_cycle, cycle, cycle->log);
+ }
+
+#endif
for ( ;; ) {
ngx_log_debug0(NGX_LOG_DEBUG_EVENT, cycle->log, 0, "worker cycle");
}
+#if (NGX_THREADS)
+
+int ngx_worker_thread_cycle(void *data)
+{
+ ngx_cycle_t *cycle = data;
+
+ struct timeval tv;
+
+ /* STUB */
+
+ ngx_log_debug1(NGX_LOG_DEBUG_CORE, ngx_cycle->log, ngx_errno,
+ "thread %d started", ngx_thread_self());
+
+ ngx_setproctitle("worker thread");
+
+ sleep(5);
+
+ ngx_gettimeofday(&tv);
+ ngx_time_update(tv.tv_sec);
+
+ ngx_log_debug1(NGX_LOG_DEBUG_CORE, ngx_cycle->log, ngx_errno,
+ "thread %d done", ngx_thread_self());
+
+ return 1;
+}
+
+#endif
+
+
static ngx_int_t ngx_add_inherited_sockets(ngx_cycle_t *cycle, char **envp)
{
char *p, *v;
#include <ngx_config.h>
#include <ngx_core.h>
+/*
+ * 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.
+ */
-extern int __isthreaded;
-
-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()
{
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;
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);
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 */
+
+ if (!(errnos = ngx_calloc(n * sizeof(int), log))) {
+ return NGX_ERROR;
+ }
+
+ /* create the threads tid array */
- /* create the thread tid array */
- ngx_test_null(tids, ngx_calloc((n + 1) * sizeof(ngx_tid_t), log),
- NGX_ERROR);
+ if (!(tids = ngx_calloc((n + 1) * sizeof(ngx_tid_t), log))) {
+ return NGX_ERROR;
+ }
- tids[0] = ngx_getpid();
- threads = 1;
+ 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;
}
+static inline int ngx_gettid()
+{
+ char *sp;
+
+ if (stack_size == 0) {
+ return 0;
+ }
+
+ __asm__ ("mov %%esp, %0" : "=q" (sp));
+
+ return (usrstack - sp) / stack_size;
+}
+
+
ngx_tid_t ngx_thread_self()
{
int tid;
tid = ngx_gettid();
+ if (tids == NULL) {
+ return ngx_pid;
+ }
+
+#if 0
if (tids[tid] == 0) {
- pid = ngx_getpid();
+ pid = ngx_pid;
tids[tid] = pid;
return pid;
}
+#endif
return tids[tid];
}
-static inline int ngx_gettid()
+ngx_mutex_t *ngx_mutex_init(ngx_log_t *log, uint flags)
{
- char *sp;
+ int nsem, i;
+ ngx_mutex_t *m;
+ union semun op;
- if (stack_size == 0) {
- return 0;
+ if (!(m = ngx_alloc(sizeof(ngx_mutex_t), log))) {
+ return NULL;
}
- __asm__ ("mov %%esp, %0" : "=q" (sp));
+ m->lock = 0;
+ m->log = log;
- return (usrstack - sp) / stack_size;
+ 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;
}
projects / nginx.git / commitdiff