Extend notifier_call_chain to count nr_calls made

[linux-2.6] / fs / dlm / requestqueue.c

diff --git a/fs/dlm/requestqueue.c b/fs/dlm/requestqueue.c
index 36afe99e4f93e1e62386700f57c9739c55ac5b33..65008d79c96d2e651091dbd7bcbe47f1e17221ec 100644 (file)
--- a/fs/dlm/requestqueue.c
+++ b/fs/dlm/requestqueue.c
@@ -30,26 +30,36 @@ struct rq_entry {
  * lockspace is enabled on some while still suspended on others.
  */
 
-void dlm_add_requestqueue(struct dlm_ls *ls, int nodeid, struct dlm_header *hd)
+int dlm_add_requestqueue(struct dlm_ls *ls, int nodeid, struct dlm_header *hd)
 {
        struct rq_entry *e;
        int length = hd->h_length;
-
-       if (dlm_is_removed(ls, nodeid))
-               return;
+       int rv = 0;
 
        e = kmalloc(sizeof(struct rq_entry) + length, GFP_KERNEL);
        if (!e) {
                log_print("dlm_add_requestqueue: out of memory\n");
-               return;
+               return 0;
        }
 
        e->nodeid = nodeid;
        memcpy(e->request, hd, length);
 
-       down(&ls->ls_requestqueue_lock);
-       list_add_tail(&e->list, &ls->ls_requestqueue);
-       up(&ls->ls_requestqueue_lock);
+       /* We need to check dlm_locking_stopped() after taking the mutex to
+          avoid a race where dlm_recoverd enables locking and runs
+          process_requestqueue between our earlier dlm_locking_stopped check
+          and this addition to the requestqueue. */
+
+       mutex_lock(&ls->ls_requestqueue_mutex);
+       if (dlm_locking_stopped(ls))
+               list_add_tail(&e->list, &ls->ls_requestqueue);
+       else {
+               log_debug(ls, "dlm_add_requestqueue skip from %d", nodeid);
+               kfree(e);
+               rv = -EAGAIN;
+       }
+       mutex_unlock(&ls->ls_requestqueue_mutex);
+       return rv;
 }
 
 int dlm_process_requestqueue(struct dlm_ls *ls)
@@ -58,19 +68,19 @@ int dlm_process_requestqueue(struct dlm_ls *ls)
        struct dlm_header *hd;
        int error = 0;
 
-       down(&ls->ls_requestqueue_lock);
+       mutex_lock(&ls->ls_requestqueue_mutex);
 
        for (;;) {
                if (list_empty(&ls->ls_requestqueue)) {
-                       up(&ls->ls_requestqueue_lock);
+                       mutex_unlock(&ls->ls_requestqueue_mutex);
                        error = 0;
                        break;
                }
                e = list_entry(ls->ls_requestqueue.next, struct rq_entry, list);
-               up(&ls->ls_requestqueue_lock);
+               mutex_unlock(&ls->ls_requestqueue_mutex);
 
                hd = (struct dlm_header *) e->request;
-               error = dlm_receive_message(hd, e->nodeid, TRUE);
+               error = dlm_receive_message(hd, e->nodeid, 1);
 
                if (error == -EINTR) {
                        /* entry is left on requestqueue */
@@ -78,13 +88,13 @@ int dlm_process_requestqueue(struct dlm_ls *ls)
                        break;
                }
 
-               down(&ls->ls_requestqueue_lock);
+               mutex_lock(&ls->ls_requestqueue_mutex);
                list_del(&e->list);
                kfree(e);
 
                if (dlm_locking_stopped(ls)) {
                        log_debug(ls, "process_requestqueue abort running");
-                       up(&ls->ls_requestqueue_lock);
+                       mutex_unlock(&ls->ls_requestqueue_mutex);
                        error = -EINTR;
                        break;
                }
@@ -105,21 +115,25 @@ int dlm_process_requestqueue(struct dlm_ls *ls)
 void dlm_wait_requestqueue(struct dlm_ls *ls)
 {
        for (;;) {
-               down(&ls->ls_requestqueue_lock);
+               mutex_lock(&ls->ls_requestqueue_mutex);
                if (list_empty(&ls->ls_requestqueue))
                        break;
                if (dlm_locking_stopped(ls))
                        break;
-               up(&ls->ls_requestqueue_lock);
+               mutex_unlock(&ls->ls_requestqueue_mutex);
                schedule();
        }
-       up(&ls->ls_requestqueue_lock);
+       mutex_unlock(&ls->ls_requestqueue_mutex);
 }
 
 static int purge_request(struct dlm_ls *ls, struct dlm_message *ms, int nodeid)
 {
        uint32_t type = ms->m_type;
 
+       /* the ls is being cleaned up and freed by release_lockspace */
+       if (!ls->ls_count)
+               return 1;
+
        if (dlm_is_removed(ls, nodeid))
                return 1;
 
@@ -170,7 +184,7 @@ void dlm_purge_requestqueue(struct dlm_ls *ls)
        struct dlm_message *ms;
        struct rq_entry *e, *safe;
 
-       down(&ls->ls_requestqueue_lock);
+       mutex_lock(&ls->ls_requestqueue_mutex);
        list_for_each_entry_safe(e, safe, &ls->ls_requestqueue, list) {
                ms = (struct dlm_message *) e->request;
 
@@ -179,6 +193,6 @@ void dlm_purge_requestqueue(struct dlm_ls *ls)
                        kfree(e);
                }
        }
-       up(&ls->ls_requestqueue_lock);
+       mutex_unlock(&ls->ls_requestqueue_mutex);
 }