Mercurial > hg > nginx

changeset 4309:3f6040cd731e

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Added shmtx interface to forcibly unlock mutexes. It is currently used from master process on abnormal worker termination to unlock accept mutex (unlocking of accept mutex was broken in 1.0.2). It is expected to be used in the future to unlock other mutexes as well. Shared mutex code was rewritten to make this possible in a safe way, i.e. with a check if lock was actually held by the exited process. We again use pid to lock mutex, and use separate atomic variable for a count of processes waiting in sem_wait().
author Maxim Dounin <mdounin@mdounin.ru>
date Wed, 23 Nov 2011 13:55:38 +0000
parents 39982fa69482
children 13f108b9f3cf
files src/core/ngx_cycle.c src/core/ngx_shmtx.c src/core/ngx_shmtx.h src/core/ngx_slab.h src/event/ngx_event.c src/os/unix/ngx_process.c
diffstat 6 files changed, 88 insertions(+), 51 deletions(-) [+]
line wrap: on
line diff
--- a/src/core/ngx_cycle.c
+++ b/src/core/ngx_cycle.c
@@ -952,7 +952,7 @@ ngx_init_zone_pool(ngx_cycle_t *cycle, n
 
 #endif
 
-    if (ngx_shmtx_create(&sp->mutex, (void *) &sp->lock, file) != NGX_OK) {
+    if (ngx_shmtx_create(&sp->mutex, &sp->lock, file) != NGX_OK) {
         return NGX_ERROR;
     }
 
--- a/src/core/ngx_shmtx.c
+++ b/src/core/ngx_shmtx.c
@@ -11,10 +11,13 @@
 #if (NGX_HAVE_ATOMIC_OPS)
 
 
+static void ngx_shmtx_wakeup(ngx_shmtx_t *mtx);
+
+
 ngx_int_t
-ngx_shmtx_create(ngx_shmtx_t *mtx, void *addr, u_char *name)
+ngx_shmtx_create(ngx_shmtx_t *mtx, ngx_shmtx_sh_t *addr, u_char *name)
 {
-    mtx->lock = addr;
+    mtx->lock = &addr->lock;
 
     if (mtx->spin == (ngx_uint_t) -1) {
         return NGX_OK;
@@ -24,6 +27,8 @@ ngx_shmtx_create(ngx_shmtx_t *mtx, void 
 
 #if (NGX_HAVE_POSIX_SEM)
 
+    mtx->wait = &addr->wait;
+
     if (sem_init(&mtx->sem, 1, 0) == -1) {
         ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, ngx_errno,
                       "sem_init() failed");
@@ -56,12 +61,7 @@ ngx_shmtx_destory(ngx_shmtx_t *mtx)
 ngx_uint_t
 ngx_shmtx_trylock(ngx_shmtx_t *mtx)
 {
-    ngx_atomic_uint_t  val;
-
-    val = *mtx->lock;
-
-    return ((val & 0x80000000) == 0
-            && ngx_atomic_cmp_set(mtx->lock, val, val | 0x80000000));
+    return (*mtx->lock == 0 && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid));
 }
 
 
@@ -69,17 +69,12 @@ void
 ngx_shmtx_lock(ngx_shmtx_t *mtx)
 {
     ngx_uint_t         i, n;
-    ngx_atomic_uint_t  val;
 
     ngx_log_debug0(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0, "shmtx lock");
 
     for ( ;; ) {
 
-        val = *mtx->lock;
-
-        if ((val & 0x80000000) == 0
-            && ngx_atomic_cmp_set(mtx->lock, val, val | 0x80000000))
-        {
+        if (*mtx->lock == 0 && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid)) {
             return;
         }
 
@@ -91,10 +86,8 @@ ngx_shmtx_lock(ngx_shmtx_t *mtx)
                     ngx_cpu_pause();
                 }
 
-                val = *mtx->lock;
-
-                if ((val & 0x80000000) == 0
-                    && ngx_atomic_cmp_set(mtx->lock, val, val | 0x80000000))
+                if (*mtx->lock == 0
+                    && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid))
                 {
                     return;
                 }
@@ -104,24 +97,24 @@ ngx_shmtx_lock(ngx_shmtx_t *mtx)
 #if (NGX_HAVE_POSIX_SEM)
 
         if (mtx->semaphore) {
-            val = *mtx->lock;
+            ngx_atomic_fetch_add(mtx->wait, 1);
 
-            if ((val & 0x80000000)
-                && ngx_atomic_cmp_set(mtx->lock, val, val + 1))
-            {
-                ngx_log_debug1(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0,
-                               "shmtx wait %XA", val);
+            if (*mtx->lock == 0 && ngx_atomic_cmp_set(mtx->lock, 0, ngx_pid)) {
+                return;
+            }
+
+            ngx_log_debug1(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0,
+                           "shmtx wait %uA", *mtx->wait);
 
-                while (sem_wait(&mtx->sem) == -1) {
-                    ngx_err_t  err;
+            while (sem_wait(&mtx->sem) == -1) {
+                ngx_err_t  err;
 
-                    err = ngx_errno;
+                err = ngx_errno;
 
-                    if (err != NGX_EINTR) {
-                        ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, err,
-                                   "sem_wait() failed while waiting on shmtx");
-                        break;
-                    }
+                if (err != NGX_EINTR) {
+                    ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, err,
+                                  "sem_wait() failed while waiting on shmtx");
+                    break;
                 }
 
                 ngx_log_debug0(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0,
@@ -141,31 +134,56 @@ ngx_shmtx_lock(ngx_shmtx_t *mtx)
 void
 ngx_shmtx_unlock(ngx_shmtx_t *mtx)
 {
-    ngx_atomic_uint_t  val, old, wait;
-
     if (mtx->spin != (ngx_uint_t) -1) {
         ngx_log_debug0(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0, "shmtx unlock");
     }
 
+    if (ngx_atomic_cmp_set(mtx->lock, ngx_pid, 0)) {
+        ngx_shmtx_wakeup(mtx);
+    }
+}
+
+
+ngx_uint_t
+ngx_shmtx_force_unlock(ngx_shmtx_t *mtx, ngx_pid_t pid)
+{
+    ngx_log_debug0(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0,
+                   "shmtx forced unlock");
+
+    if (ngx_atomic_cmp_set(mtx->lock, pid, 0)) {
+        ngx_shmtx_wakeup(mtx);
+        return 1;
+    }
+
+    return 0;
+}
+
+
+static void
+ngx_shmtx_wakeup(ngx_shmtx_t *mtx)
+{
+#if (NGX_HAVE_POSIX_SEM)
+    ngx_atomic_uint_t  wait;
+
+    if (!mtx->semaphore) {
+        return;
+    }
+
     for ( ;; ) {
 
-        old = *mtx->lock;
-        wait = old & 0x7fffffff;
-        val = wait ? wait - 1 : 0;
+        wait = *mtx->wait;
 
-        if (ngx_atomic_cmp_set(mtx->lock, old, val)) {
+        if (wait == 0) {
+            return;
+        }
+
+        if (ngx_atomic_cmp_set(mtx->wait, wait, wait - 1)) {
             break;
         }
     }
 
-#if (NGX_HAVE_POSIX_SEM)
-
-    if (wait == 0 || !mtx->semaphore) {
-        return;
-    }
-
     ngx_log_debug1(NGX_LOG_DEBUG_CORE, ngx_cycle->log, 0,
-                   "shmtx wake %XA", old);
+                   "shmtx wake %uA", wait);
 
     if (sem_post(&mtx->sem) == -1) {
         ngx_log_error(NGX_LOG_ALERT, ngx_cycle->log, ngx_errno,
@@ -180,7 +198,7 @@ ngx_shmtx_unlock(ngx_shmtx_t *mtx)
 
 
 ngx_int_t
-ngx_shmtx_create(ngx_shmtx_t *mtx, void *addr, u_char *name)
+ngx_shmtx_create(ngx_shmtx_t *mtx, ngx_shmtx_sh_t *addr, u_char *name)
 {
     if (mtx->name) {
 
@@ -280,4 +298,11 @@ ngx_shmtx_unlock(ngx_shmtx_t *mtx)
     ngx_log_abort(err, ngx_unlock_fd_n " %s failed", mtx->name);
 }
 
+
+void
+ngx_shmtx_force_unlock(ngx_shmtx_t *mtx, ngx_pid_t pid)
+{
+    /* void */
+}
+
 #endif
--- a/src/core/ngx_shmtx.h
+++ b/src/core/ngx_shmtx.h
@@ -13,9 +13,18 @@
 
 
 typedef struct {
+    ngx_atomic_t   lock;
+#if (NGX_HAVE_POSIX_SEM)
+    ngx_atomic_t   wait;
+#endif
+} ngx_shmtx_sh_t;
+
+
+typedef struct {
 #if (NGX_HAVE_ATOMIC_OPS)
     ngx_atomic_t  *lock;
 #if (NGX_HAVE_POSIX_SEM)
+    ngx_atomic_t  *wait;
     ngx_uint_t     semaphore;
     sem_t          sem;
 #endif
@@ -27,11 +36,13 @@ typedef struct {
 } ngx_shmtx_t;
 
 
-ngx_int_t ngx_shmtx_create(ngx_shmtx_t *mtx, void *addr, u_char *name);
+ngx_int_t ngx_shmtx_create(ngx_shmtx_t *mtx, ngx_shmtx_sh_t *addr,
+    u_char *name);
 void ngx_shmtx_destory(ngx_shmtx_t *mtx);
 ngx_uint_t ngx_shmtx_trylock(ngx_shmtx_t *mtx);
 void ngx_shmtx_lock(ngx_shmtx_t *mtx);
 void ngx_shmtx_unlock(ngx_shmtx_t *mtx);
+ngx_uint_t ngx_shmtx_force_unlock(ngx_shmtx_t *mtx, ngx_pid_t pid);
 
 
 #endif /* _NGX_SHMTX_H_INCLUDED_ */
--- a/src/core/ngx_slab.h
+++ b/src/core/ngx_slab.h
@@ -22,7 +22,7 @@ struct ngx_slab_page_s {
 
 
 typedef struct {
-    ngx_atomic_t      lock;
+    ngx_shmtx_sh_t    lock;
 
     size_t            min_size;
     size_t            min_shift;
--- a/src/event/ngx_event.c
+++ b/src/event/ngx_event.c
@@ -521,7 +521,8 @@ ngx_event_module_init(ngx_cycle_t *cycle
     ngx_accept_mutex_ptr = (ngx_atomic_t *) shared;
     ngx_accept_mutex.spin = (ngx_uint_t) -1;
 
-    if (ngx_shmtx_create(&ngx_accept_mutex, shared, cycle->lock_file.data)
+    if (ngx_shmtx_create(&ngx_accept_mutex, (ngx_shmtx_sh_t *) shared,
+                         cycle->lock_file.data)
         != NGX_OK)
     {
         return NGX_ERROR;
--- a/src/os/unix/ngx_process.c
+++ b/src/os/unix/ngx_process.c
@@ -504,7 +504,7 @@ ngx_process_get_status(void)
              * held it
              */
 
-            ngx_atomic_cmp_set(ngx_accept_mutex_ptr, pid, 0);
+            ngx_shmtx_force_unlock(&ngx_accept_mutex, pid);
         }