17 files changed, 236 insertions, 114 deletions

diff --git a/fs/dlm/ast.c b/fs/dlm/ast.c
index dc2ad6008b2d..4314f0d48d85 100644
--- a/fs/dlm/ast.c
+++ b/fs/dlm/ast.c
@@ -2,7 +2,7 @@
 *******************************************************************************
 **
 **  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
-**  Copyright (C) 2004-2008 Red Hat, Inc.  All rights reserved.
+**  Copyright (C) 2004-2010 Red Hat, Inc.  All rights reserved.
 **
 **  This copyrighted material is made available to anyone wishing to use,
 **  modify, copy, or redistribute it subject to the terms and conditions
@@ -33,10 +33,10 @@ void dlm_del_ast(struct dlm_lkb *lkb)
         spin_unlock(&ast_queue_lock);
 }
 
-void dlm_add_ast(struct dlm_lkb *lkb, int type, int bastmode)
+void dlm_add_ast(struct dlm_lkb *lkb, int type, int mode)
 {
         if (lkb->lkb_flags & DLM_IFL_USER) {
-                dlm_user_add_ast(lkb, type, bastmode);
+                dlm_user_add_ast(lkb, type, mode);
                 return;
         }
 
@@ -44,10 +44,21 @@ void dlm_add_ast(struct dlm_lkb *lkb, int type, int bastmode)
         if (!(lkb->lkb_ast_type & (AST_COMP | AST_BAST))) {
                 kref_get(&lkb->lkb_ref);
                 list_add_tail(&lkb->lkb_astqueue, &ast_queue);
+                lkb->lkb_ast_first = type;
         }
+
+        /* sanity check, this should not happen */
+
+        if ((type == AST_COMP) && (lkb->lkb_ast_type & AST_COMP))
+                log_print("repeat cast %d castmode %d lock %x %s",
+                          mode, lkb->lkb_castmode,
+                          lkb->lkb_id, lkb->lkb_resource->res_name);
+
         lkb->lkb_ast_type |= type;
-        if (bastmode)
-                lkb->lkb_bastmode = bastmode;
+        if (type == AST_BAST)
+                lkb->lkb_bastmode = mode;
+        else
+                lkb->lkb_castmode = mode;
         spin_unlock(&ast_queue_lock);
 
         set_bit(WAKE_ASTS, &astd_wakeflags);
@@ -59,9 +70,9 @@ static void process_asts(void)
         struct dlm_ls *ls = NULL;
         struct dlm_rsb *r = NULL;
         struct dlm_lkb *lkb;
-        void (*cast) (void *astparam);
-        void (*bast) (void *astparam, int mode);
-        int type = 0, bastmode;
+        void (*castfn) (void *astparam);
+        void (*bastfn) (void *astparam, int mode);
+        int type, first, bastmode, castmode, do_bast, do_cast, last_castmode;
 
 repeat:
         spin_lock(&ast_queue_lock);
@@ -75,17 +86,48 @@ repeat:
                 list_del(&lkb->lkb_astqueue);
                 type = lkb->lkb_ast_type;
                 lkb->lkb_ast_type = 0;
+                first = lkb->lkb_ast_first;
+                lkb->lkb_ast_first = 0;
                 bastmode = lkb->lkb_bastmode;
-
+                castmode = lkb->lkb_castmode;
+                castfn = lkb->lkb_astfn;
+                bastfn = lkb->lkb_bastfn;
                 spin_unlock(&ast_queue_lock);
-                cast = lkb->lkb_astfn;
-                bast = lkb->lkb_bastfn;
-
-                if ((type & AST_COMP) && cast)
-                        cast(lkb->lkb_astparam);
 
-                if ((type & AST_BAST) && bast)
-                        bast(lkb->lkb_astparam, bastmode);
+                do_cast = (type & AST_COMP) && castfn;
+                do_bast = (type & AST_BAST) && bastfn;
+
+                /* Skip a bast if its blocking mode is compatible with the
+                   granted mode of the preceding cast. */
+
+                if (do_bast) {
+                        if (first == AST_COMP)
+                                last_castmode = castmode;
+                        else
+                                last_castmode = lkb->lkb_castmode_done;
+                        if (dlm_modes_compat(bastmode, last_castmode))
+                                do_bast = 0;
+                }
+
+                if (first == AST_COMP) {
+                        if (do_cast)
+                                castfn(lkb->lkb_astparam);
+                        if (do_bast)
+                                bastfn(lkb->lkb_astparam, bastmode);
+                } else if (first == AST_BAST) {
+                        if (do_bast)
+                                bastfn(lkb->lkb_astparam, bastmode);
+                        if (do_cast)
+                                castfn(lkb->lkb_astparam);
+                } else {
+                        log_error(ls, "bad ast_first %d ast_type %d",
+                                  first, type);
+                }
+
+                if (do_cast)
+                        lkb->lkb_castmode_done = castmode;
+                if (do_bast)
+                        lkb->lkb_bastmode_done = bastmode;
 
                 /* this removes the reference added by dlm_add_ast
                    and may result in the lkb being freed */
diff --git a/fs/dlm/ast.h b/fs/dlm/ast.h
index 1b5fc5f428fd..bcb1aaba519d 100644
--- a/fs/dlm/ast.h
+++ b/fs/dlm/ast.h
@@ -1,7 +1,7 @@
 /******************************************************************************
 *******************************************************************************
 **
-**  Copyright (C) 2005-2008 Red Hat, Inc.  All rights reserved.
+**  Copyright (C) 2005-2010 Red Hat, Inc.  All rights reserved.
 **
 **  This copyrighted material is made available to anyone wishing to use,
 **  modify, copy, or redistribute it subject to the terms and conditions
@@ -13,7 +13,7 @@
 #ifndef __ASTD_DOT_H__
 #define __ASTD_DOT_H__
 
-void dlm_add_ast(struct dlm_lkb *lkb, int type, int bastmode);
+void dlm_add_ast(struct dlm_lkb *lkb, int type, int mode);
 void dlm_del_ast(struct dlm_lkb *lkb);
 
 void dlm_astd_wake(void);
diff --git a/fs/dlm/config.c b/fs/dlm/config.c
index fd9859f92fad..b54bca03d92f 100644
--- a/fs/dlm/config.c
+++ b/fs/dlm/config.c
@@ -14,6 +14,7 @@
 #include <linux/kernel.h>
 #include <linux/module.h>
 #include <linux/configfs.h>
+#include <linux/slab.h>
 #include <linux/in.h>
 #include <linux/in6.h>
 #include <net/ipv6.h>
@@ -410,10 +411,10 @@ static struct config_group *make_cluster(struct config_group *g,
         struct dlm_comms *cms = NULL;
         void *gps = NULL;
 
-        cl = kzalloc(sizeof(struct dlm_cluster), GFP_KERNEL);
-        gps = kcalloc(3, sizeof(struct config_group *), GFP_KERNEL);
-        sps = kzalloc(sizeof(struct dlm_spaces), GFP_KERNEL);
-        cms = kzalloc(sizeof(struct dlm_comms), GFP_KERNEL);
+        cl = kzalloc(sizeof(struct dlm_cluster), GFP_NOFS);
+        gps = kcalloc(3, sizeof(struct config_group *), GFP_NOFS);
+        sps = kzalloc(sizeof(struct dlm_spaces), GFP_NOFS);
+        cms = kzalloc(sizeof(struct dlm_comms), GFP_NOFS);
 
         if (!cl || !gps || !sps || !cms)
                 goto fail;
@@ -482,9 +483,9 @@ static struct config_group *make_space(struct config_group *g, const char *name)
         struct dlm_nodes *nds = NULL;
         void *gps = NULL;
 
-        sp = kzalloc(sizeof(struct dlm_space), GFP_KERNEL);
-        gps = kcalloc(2, sizeof(struct config_group *), GFP_KERNEL);
-        nds = kzalloc(sizeof(struct dlm_nodes), GFP_KERNEL);
+        sp = kzalloc(sizeof(struct dlm_space), GFP_NOFS);
+        gps = kcalloc(2, sizeof(struct config_group *), GFP_NOFS);
+        nds = kzalloc(sizeof(struct dlm_nodes), GFP_NOFS);
 
         if (!sp || !gps || !nds)
                 goto fail;
@@ -536,7 +537,7 @@ static struct config_item *make_comm(struct config_group *g, const char *name)
 {
         struct dlm_comm *cm;
 
-        cm = kzalloc(sizeof(struct dlm_comm), GFP_KERNEL);
+        cm = kzalloc(sizeof(struct dlm_comm), GFP_NOFS);
         if (!cm)
                 return ERR_PTR(-ENOMEM);
 
@@ -569,7 +570,7 @@ static struct config_item *make_node(struct config_group *g, const char *name)
         struct dlm_space *sp = config_item_to_space(g->cg_item.ci_parent);
         struct dlm_node *nd;
 
-        nd = kzalloc(sizeof(struct dlm_node), GFP_KERNEL);
+        nd = kzalloc(sizeof(struct dlm_node), GFP_NOFS);
         if (!nd)
                 return ERR_PTR(-ENOMEM);
 
@@ -705,7 +706,7 @@ static ssize_t comm_addr_write(struct dlm_comm *cm, const char *buf, size_t len)
         if (cm->addr_count >= DLM_MAX_ADDR_COUNT)
                 return -ENOSPC;
 
-        addr = kzalloc(sizeof(*addr), GFP_KERNEL);
+        addr = kzalloc(sizeof(*addr), GFP_NOFS);
         if (!addr)
                 return -ENOMEM;
 
@@ -868,7 +869,7 @@ int dlm_nodeid_list(char *lsname, int **ids_out, int *ids_count_out,
 
         ids_count = sp->members_count;
 
-        ids = kcalloc(ids_count, sizeof(int), GFP_KERNEL);
+        ids = kcalloc(ids_count, sizeof(int), GFP_NOFS);
         if (!ids) {
                 rv = -ENOMEM;
                 goto out;
@@ -886,7 +887,7 @@ int dlm_nodeid_list(char *lsname, int **ids_out, int *ids_count_out,
         if (!new_count)
                 goto out_ids;
 
-        new = kcalloc(new_count, sizeof(int), GFP_KERNEL);
+        new = kcalloc(new_count, sizeof(int), GFP_NOFS);
         if (!new) {
                 kfree(ids);
                 rv = -ENOMEM;
diff --git a/fs/dlm/debug_fs.c b/fs/dlm/debug_fs.c
index 1c8bb8c3a82e..c6cf25158746 100644
--- a/fs/dlm/debug_fs.c
+++ b/fs/dlm/debug_fs.c
@@ -15,6 +15,7 @@
 #include <linux/module.h>
 #include <linux/ctype.h>
 #include <linux/debugfs.h>
+#include <linux/slab.h>
 
 #include "dlm_internal.h"
 #include "lock.h"
@@ -256,7 +257,7 @@ static int print_format3_lock(struct seq_file *s, struct dlm_lkb *lkb,
                         lkb->lkb_status,
                         lkb->lkb_grmode,
                         lkb->lkb_rqmode,
-                        lkb->lkb_highbast,
+                        lkb->lkb_bastmode,
                         rsb_lookup,
                         lkb->lkb_wait_type,
                         lkb->lkb_lvbseq,
@@ -404,7 +405,7 @@ static void *table_seq_start(struct seq_file *seq, loff_t *pos)
         if (bucket >= ls->ls_rsbtbl_size)
                 return NULL;
 
-        ri = kzalloc(sizeof(struct rsbtbl_iter), GFP_KERNEL);
+        ri = kzalloc(sizeof(struct rsbtbl_iter), GFP_NOFS);
         if (!ri)
                 return NULL;
         if (n == 0)
diff --git a/fs/dlm/dir.c b/fs/dlm/dir.c
index c4dfa1dcc86f..7b84c1dbc82e 100644
--- a/fs/dlm/dir.c
+++ b/fs/dlm/dir.c
@@ -49,8 +49,7 @@ static struct dlm_direntry *get_free_de(struct dlm_ls *ls, int len)
         spin_unlock(&ls->ls_recover_list_lock);
 
         if (!found)
-                de = kzalloc(sizeof(struct dlm_direntry) + len,
-                             ls->ls_allocation);
+                de = kzalloc(sizeof(struct dlm_direntry) + len, GFP_NOFS);
         return de;
 }
 
@@ -212,7 +211,7 @@ int dlm_recover_directory(struct dlm_ls *ls)
 
         dlm_dir_clear(ls);
 
-        last_name = kmalloc(DLM_RESNAME_MAXLEN, ls->ls_allocation);
+        last_name = kmalloc(DLM_RESNAME_MAXLEN, GFP_NOFS);
         if (!last_name)
                 goto out;
 
@@ -323,7 +322,7 @@ static int get_entry(struct dlm_ls *ls, int nodeid, char *name,
         if (namelen > DLM_RESNAME_MAXLEN)
                 return -EINVAL;
 
-        de = kzalloc(sizeof(struct dlm_direntry) + namelen, ls->ls_allocation);
+        de = kzalloc(sizeof(struct dlm_direntry) + namelen, GFP_NOFS);
         if (!de)
                 return -ENOMEM;
 
diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h
index d01ca0a711db..f632b58cd222 100644
--- a/fs/dlm/dlm_internal.h
+++ b/fs/dlm/dlm_internal.h
@@ -2,7 +2,7 @@
 *******************************************************************************
 **
 **  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
-**  Copyright (C) 2004-2008 Red Hat, Inc.  All rights reserved.
+**  Copyright (C) 2004-2010 Red Hat, Inc.  All rights reserved.
 **
 **  This copyrighted material is made available to anyone wishing to use,
 **  modify, copy, or redistribute it subject to the terms and conditions
@@ -232,11 +232,17 @@ struct dlm_lkb {
         int8_t                  lkb_status;     /* granted, waiting, convert */
         int8_t                  lkb_rqmode;     /* requested lock mode */
         int8_t                  lkb_grmode;     /* granted lock mode */
-        int8_t                  lkb_bastmode;   /* requested mode */
         int8_t                  lkb_highbast;   /* highest mode bast sent for */
+
         int8_t                  lkb_wait_type;  /* type of reply waiting for */
         int8_t                  lkb_wait_count;
         int8_t                  lkb_ast_type;   /* type of ast queued for */
+        int8_t                  lkb_ast_first;  /* type of first ast queued */
+
+        int8_t                  lkb_bastmode;   /* req mode of queued bast */
+        int8_t                  lkb_castmode;   /* gr mode of queued cast */
+        int8_t                  lkb_bastmode_done; /* last delivered bastmode */
+        int8_t                  lkb_castmode_done; /* last delivered castmode */
 
         struct list_head        lkb_idtbl_list; /* lockspace lkbtbl */
         struct list_head        lkb_statequeue; /* rsb g/c/w list */
@@ -473,7 +479,6 @@ struct dlm_ls {
         int                     ls_low_nodeid;
         int                     ls_total_weight;
         int                     *ls_node_array;
-        gfp_t                   ls_allocation;
 
         struct dlm_rsb          ls_stub_rsb;    /* for returning errors */
         struct dlm_lkb          ls_stub_lkb;    /* for returning errors */
diff --git a/fs/dlm/lock.c b/fs/dlm/lock.c
index eb507c453c5f..17903b491298 100644
--- a/fs/dlm/lock.c
+++ b/fs/dlm/lock.c
@@ -1,7 +1,7 @@
 /******************************************************************************
 *******************************************************************************
 **
-**  Copyright (C) 2005-2008 Red Hat, Inc.  All rights reserved.
+**  Copyright (C) 2005-2010 Red Hat, Inc.  All rights reserved.
 **
 **  This copyrighted material is made available to anyone wishing to use,
 **  modify, copy, or redistribute it subject to the terms and conditions
@@ -56,6 +56,7 @@
    L: receive_xxxx_reply()     <-  R: send_xxxx_reply()
 */
 #include <linux/types.h>
+#include <linux/slab.h>
 #include "dlm_internal.h"
 #include <linux/dlm_device.h>
 #include "memory.h"
@@ -307,7 +308,7 @@ static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
         lkb->lkb_lksb->sb_status = rv;
         lkb->lkb_lksb->sb_flags = lkb->lkb_sbflags;
 
-        dlm_add_ast(lkb, AST_COMP, 0);
+        dlm_add_ast(lkb, AST_COMP, lkb->lkb_grmode);
 }
 
 static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
@@ -320,10 +321,12 @@ static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
 {
         lkb->lkb_time_bast = ktime_get();
 
-        if (is_master_copy(lkb))
+        if (is_master_copy(lkb)) {
+                lkb->lkb_bastmode = rqmode; /* printed by debugfs */
                 send_bast(r, lkb, rqmode);
-        else
+        } else {
                 dlm_add_ast(lkb, AST_BAST, rqmode);
+        }
 }
 
 /*
@@ -2280,20 +2283,30 @@ static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
         if (can_be_queued(lkb)) {
                 error = -EINPROGRESS;
                 add_lkb(r, lkb, DLM_LKSTS_WAITING);
-                send_blocking_asts(r, lkb);
                 add_timeout(lkb);
                 goto out;
         }
 
         error = -EAGAIN;
-        if (force_blocking_asts(lkb))
-                send_blocking_asts_all(r, lkb);
         queue_cast(r, lkb, -EAGAIN);
-
  out:
         return error;
 }
 
+static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
+                               int error)
+{
+        switch (error) {
+        case -EAGAIN:
+                if (force_blocking_asts(lkb))
+                        send_blocking_asts_all(r, lkb);
+                break;
+        case -EINPROGRESS:
+                send_blocking_asts(r, lkb);
+                break;
+        }
+}
+
 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
 {
         int error = 0;
@@ -2304,7 +2317,6 @@ static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
         if (can_be_granted(r, lkb, 1, &deadlk)) {
                 grant_lock(r, lkb);
                 queue_cast(r, lkb, 0);
-                grant_pending_locks(r);
                 goto out;
         }
 
@@ -2334,7 +2346,6 @@ static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
                 if (_can_be_granted(r, lkb, 1)) {
                         grant_lock(r, lkb);
                         queue_cast(r, lkb, 0);
-                        grant_pending_locks(r);
                         goto out;
                 }
                 /* else fall through and move to convert queue */
@@ -2344,28 +2355,47 @@ static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
                 error = -EINPROGRESS;
                 del_lkb(r, lkb);
                 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
-                send_blocking_asts(r, lkb);
                 add_timeout(lkb);
                 goto out;
         }
 
         error = -EAGAIN;
-        if (force_blocking_asts(lkb))
-                send_blocking_asts_all(r, lkb);
         queue_cast(r, lkb, -EAGAIN);
-
  out:
         return error;
 }
 
+static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
+                               int error)
+{
+        switch (error) {
+        case 0:
+                grant_pending_locks(r);
+                /* grant_pending_locks also sends basts */
+                break;
+        case -EAGAIN:
+                if (force_blocking_asts(lkb))
+                        send_blocking_asts_all(r, lkb);
+                break;
+        case -EINPROGRESS:
+                send_blocking_asts(r, lkb);
+                break;
+        }
+}
+
 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 {
         remove_lock(r, lkb);
         queue_cast(r, lkb, -DLM_EUNLOCK);
-        grant_pending_locks(r);
         return -DLM_EUNLOCK;
 }
 
+static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
+                              int error)
+{
+        grant_pending_locks(r);
+}
+
 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
  
 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
@@ -2375,12 +2405,18 @@ static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
         error = revert_lock(r, lkb);
         if (error) {
                 queue_cast(r, lkb, -DLM_ECANCEL);
-                grant_pending_locks(r);
                 return -DLM_ECANCEL;
         }
         return 0;
 }
 
+static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
+                              int error)
+{
+        if (error)
+                grant_pending_locks(r);
+}
+
 /*
  * Four stage 3 varieties:
  * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
@@ -2402,11 +2438,15 @@ static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
                 goto out;
         }
 
-        if (is_remote(r))
+        if (is_remote(r)) {
                 /* receive_request() calls do_request() on remote node */
                 error = send_request(r, lkb);
-        else
+        } else {
                 error = do_request(r, lkb);
+                /* for remote locks the request_reply is sent
+                   between do_request and do_request_effects */
+                do_request_effects(r, lkb, error);
+        }
  out:
         return error;
 }
@@ -2417,11 +2457,15 @@ static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 {
         int error;
 
-        if (is_remote(r))
+        if (is_remote(r)) {
                 /* receive_convert() calls do_convert() on remote node */
                 error = send_convert(r, lkb);
-        else
+        } else {
                 error = do_convert(r, lkb);
+                /* for remote locks the convert_reply is sent
+                   between do_convert and do_convert_effects */
+                do_convert_effects(r, lkb, error);
+        }
 
         return error;
 }
@@ -2432,11 +2476,15 @@ static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 {
         int error;
 
-        if (is_remote(r))
+        if (is_remote(r)) {
                 /* receive_unlock() calls do_unlock() on remote node */
                 error = send_unlock(r, lkb);
-        else
+        } else {
                 error = do_unlock(r, lkb);
+                /* for remote locks the unlock_reply is sent
+                   between do_unlock and do_unlock_effects */
+                do_unlock_effects(r, lkb, error);
+        }
 
         return error;
 }
@@ -2447,11 +2495,15 @@ static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 {
         int error;
 
-        if (is_remote(r))
+        if (is_remote(r)) {
                 /* receive_cancel() calls do_cancel() on remote node */
                 error = send_cancel(r, lkb);
-        else
+        } else {
                 error = do_cancel(r, lkb);
+                /* for remote locks the cancel_reply is sent
+                   between do_cancel and do_cancel_effects */
+                do_cancel_effects(r, lkb, error);
+        }
 
         return error;
 }
@@ -2689,7 +2741,7 @@ static int _create_message(struct dlm_ls *ls, int mb_len,
            pass into lowcomms_commit and a message buffer (mb) that we
            write our data into */
 
-        mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, ls->ls_allocation, &mb);
+        mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
         if (!mh)
                 return -ENOBUFS;
 
@@ -3191,6 +3243,7 @@ static void receive_request(struct dlm_ls *ls, struct dlm_message *ms)
         attach_lkb(r, lkb);
         error = do_request(r, lkb);
         send_request_reply(r, lkb, error);
+        do_request_effects(r, lkb, error);
 
         unlock_rsb(r);
         put_rsb(r);
@@ -3226,15 +3279,19 @@ static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
                 goto out;
 
         receive_flags(lkb, ms);
+
         error = receive_convert_args(ls, lkb, ms);
-        if (error)
-                goto out_reply;
+        if (error) {
+                send_convert_reply(r, lkb, error);
+                goto out;
+        }
+
         reply = !down_conversion(lkb);
 
         error = do_convert(r, lkb);
- out_reply:
         if (reply)
                 send_convert_reply(r, lkb, error);
+        do_convert_effects(r, lkb, error);
  out:
         unlock_rsb(r);
         put_rsb(r);
@@ -3266,13 +3323,16 @@ static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
                 goto out;
 
         receive_flags(lkb, ms);
+
         error = receive_unlock_args(ls, lkb, ms);
-        if (error)
-                goto out_reply;
+        if (error) {
+                send_unlock_reply(r, lkb, error);
+                goto out;
+        }
 
         error = do_unlock(r, lkb);
- out_reply:
         send_unlock_reply(r, lkb, error);
+        do_unlock_effects(r, lkb, error);
  out:
         unlock_rsb(r);
         put_rsb(r);
@@ -3307,6 +3367,7 @@ static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
 
         error = do_cancel(r, lkb);
         send_cancel_reply(r, lkb, error);
+        do_cancel_effects(r, lkb, error);
  out:
         unlock_rsb(r);
         put_rsb(r);
@@ -4512,7 +4573,7 @@ int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
         }
 
         if (flags & DLM_LKF_VALBLK) {
-                ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
+                ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
                 if (!ua->lksb.sb_lvbptr) {
                         kfree(ua);
                         __put_lkb(ls, lkb);
@@ -4582,7 +4643,7 @@ int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
         ua = lkb->lkb_ua;
 
         if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
-                ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_KERNEL);
+                ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
                 if (!ua->lksb.sb_lvbptr) {
                         error = -ENOMEM;
                         goto out_put;
diff --git a/fs/dlm/lockspace.c b/fs/dlm/lockspace.c
index d489fcc86713..f994a7dfda85 100644
--- a/fs/dlm/lockspace.c
+++ b/fs/dlm/lockspace.c
@@ -148,7 +148,7 @@ static void lockspace_kobj_release(struct kobject *k)
         kfree(ls);
 }
 
-static struct sysfs_ops dlm_attr_ops = {
+static const struct sysfs_ops dlm_attr_ops = {
         .show  = dlm_attr_show,
         .store = dlm_attr_store,
 };
@@ -191,6 +191,18 @@ static int do_uevent(struct dlm_ls *ls, int in)
         return error;
 }
 
+static int dlm_uevent(struct kset *kset, struct kobject *kobj,
+                      struct kobj_uevent_env *env)
+{
+        struct dlm_ls *ls = container_of(kobj, struct dlm_ls, ls_kobj);
+
+        add_uevent_var(env, "LOCKSPACE=%s", ls->ls_name);
+        return 0;
+}
+
+static struct kset_uevent_ops dlm_uevent_ops = {
+        .uevent = dlm_uevent,
+};
 
 int __init dlm_lockspace_init(void)
 {
@@ -199,7 +211,7 @@ int __init dlm_lockspace_init(void)
         INIT_LIST_HEAD(&lslist);
         spin_lock_init(&lslist_lock);
 
-        dlm_kset = kset_create_and_add("dlm", NULL, kernel_kobj);
+        dlm_kset = kset_create_and_add("dlm", &dlm_uevent_ops, kernel_kobj);
         if (!dlm_kset) {
                 printk(KERN_WARNING "%s: can not create kset\n", __func__);
                 return -ENOMEM;
@@ -430,7 +442,7 @@ static int new_lockspace(const char *name, int namelen, void **lockspace,
 
         error = -ENOMEM;
 
-        ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_KERNEL);
+        ls = kzalloc(sizeof(struct dlm_ls) + namelen, GFP_NOFS);
         if (!ls)
                 goto out;
         memcpy(ls->ls_name, name, namelen);
@@ -443,11 +455,6 @@ static int new_lockspace(const char *name, int namelen, void **lockspace,
         if (flags & DLM_LSFL_TIMEWARN)
                 set_bit(LSFL_TIMEWARN, &ls->ls_flags);
 
-        if (flags & DLM_LSFL_FS)
-                ls->ls_allocation = GFP_NOFS;
-        else
-                ls->ls_allocation = GFP_KERNEL;
-
         /* ls_exflags are forced to match among nodes, and we don't
            need to require all nodes to have some flags set */
         ls->ls_exflags = (flags & ~(DLM_LSFL_TIMEWARN | DLM_LSFL_FS |
@@ -456,7 +463,7 @@ static int new_lockspace(const char *name, int namelen, void **lockspace,
         size = dlm_config.ci_rsbtbl_size;
         ls->ls_rsbtbl_size = size;
 
-        ls->ls_rsbtbl = kmalloc(sizeof(struct dlm_rsbtable) * size, GFP_KERNEL);
+        ls->ls_rsbtbl = kmalloc(sizeof(struct dlm_rsbtable) * size, GFP_NOFS);
         if (!ls->ls_rsbtbl)
                 goto out_lsfree;
         for (i = 0; i < size; i++) {
@@ -468,7 +475,7 @@ static int new_lockspace(const char *name, int namelen, void **lockspace,
         size = dlm_config.ci_lkbtbl_size;
         ls->ls_lkbtbl_size = size;
 
-        ls->ls_lkbtbl = kmalloc(sizeof(struct dlm_lkbtable) * size, GFP_KERNEL);
+        ls->ls_lkbtbl = kmalloc(sizeof(struct dlm_lkbtable) * size, GFP_NOFS);
         if (!ls->ls_lkbtbl)
                 goto out_rsbfree;
         for (i = 0; i < size; i++) {
@@ -480,7 +487,7 @@ static int new_lockspace(const char *name, int namelen, void **lockspace,
         size = dlm_config.ci_dirtbl_size;
         ls->ls_dirtbl_size = size;
 
-        ls->ls_dirtbl = kmalloc(sizeof(struct dlm_dirtable) * size, GFP_KERNEL);
+        ls->ls_dirtbl = kmalloc(sizeof(struct dlm_dirtable) * size, GFP_NOFS);
         if (!ls->ls_dirtbl)
                 goto out_lkbfree;
         for (i = 0; i < size; i++) {
@@ -527,7 +534,7 @@ static int new_lockspace(const char *name, int namelen, void **lockspace,
         mutex_init(&ls->ls_requestqueue_mutex);
         mutex_init(&ls->ls_clear_proc_locks);
 
-        ls->ls_recover_buf = kmalloc(dlm_config.ci_buffer_size, GFP_KERNEL);
+        ls->ls_recover_buf = kmalloc(dlm_config.ci_buffer_size, GFP_NOFS);
         if (!ls->ls_recover_buf)
                 goto out_dirfree;
 
diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c
index 70736eb4b516..c0d35c620526 100644
--- a/fs/dlm/lowcomms.c
+++ b/fs/dlm/lowcomms.c
@@ -51,6 +51,7 @@
 #include <linux/file.h>
 #include <linux/mutex.h>
 #include <linux/sctp.h>
+#include <linux/slab.h>
 #include <net/sctp/user.h>
 #include <net/ipv6.h>
 
@@ -1060,7 +1061,7 @@ static void init_local(void)
                 if (dlm_our_addr(&sas, i))
                         break;
 
-                addr = kmalloc(sizeof(*addr), GFP_KERNEL);
+                addr = kmalloc(sizeof(*addr), GFP_NOFS);
                 if (!addr)
                         break;
                 memcpy(addr, &sas, sizeof(*addr));
@@ -1099,7 +1100,7 @@ static int sctp_listen_for_all(void)
         struct sockaddr_storage localaddr;
         struct sctp_event_subscribe subscribe;
         int result = -EINVAL, num = 1, i, addr_len;
-        struct connection *con = nodeid2con(0, GFP_KERNEL);
+        struct connection *con = nodeid2con(0, GFP_NOFS);
         int bufsize = NEEDED_RMEM;
 
         if (!con)
@@ -1171,7 +1172,7 @@ out:
 static int tcp_listen_for_all(void)
 {
         struct socket *sock = NULL;
-        struct connection *con = nodeid2con(0, GFP_KERNEL);
+        struct connection *con = nodeid2con(0, GFP_NOFS);
         int result = -EINVAL;
 
         if (!con)
diff --git a/fs/dlm/member.c b/fs/dlm/member.c
index b128775913b2..b12532e553f8 100644
--- a/fs/dlm/member.c
+++ b/fs/dlm/member.c
@@ -48,7 +48,7 @@ static int dlm_add_member(struct dlm_ls *ls, int nodeid)
         struct dlm_member *memb;
         int w, error;
 
-        memb = kzalloc(sizeof(struct dlm_member), ls->ls_allocation);
+        memb = kzalloc(sizeof(struct dlm_member), GFP_NOFS);
         if (!memb)
                 return -ENOMEM;
 
@@ -143,7 +143,7 @@ static void make_member_array(struct dlm_ls *ls)
 
         ls->ls_total_weight = total;
 
-        array = kmalloc(sizeof(int) * total, ls->ls_allocation);
+        array = kmalloc(sizeof(int) * total, GFP_NOFS);
         if (!array)
                 return;
 
@@ -226,7 +226,7 @@ int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
                         continue;
                 log_debug(ls, "new nodeid %d is a re-added member", rv->new[i]);
 
-                memb = kzalloc(sizeof(struct dlm_member), ls->ls_allocation);
+                memb = kzalloc(sizeof(struct dlm_member), GFP_NOFS);
                 if (!memb)
                         return -ENOMEM;
                 memb->nodeid = rv->new[i];
@@ -312,7 +312,7 @@ int dlm_ls_stop(struct dlm_ls *ls)
         /*
          * This in_recovery lock does two things:
          * 1) Keeps this function from returning until all threads are out
-         *    of locking routines and locking is truely stopped.
+         *    of locking routines and locking is truly stopped.
          * 2) Keeps any new requests from being processed until it's unlocked
          *    when recovery is complete.
          */
@@ -341,7 +341,7 @@ int dlm_ls_start(struct dlm_ls *ls)
         int *ids = NULL, *new = NULL;
         int error, ids_count = 0, new_count = 0;
 
-        rv = kzalloc(sizeof(struct dlm_recover), ls->ls_allocation);
+        rv = kzalloc(sizeof(struct dlm_recover), GFP_NOFS);
         if (!rv)
                 return -ENOMEM;
 
diff --git a/fs/dlm/memory.c b/fs/dlm/memory.c
index c1775b84ebab..8e0d00db004f 100644
--- a/fs/dlm/memory.c
+++ b/fs/dlm/memory.c
@@ -39,7 +39,7 @@ char *dlm_allocate_lvb(struct dlm_ls *ls)
 {
         char *p;
 
-        p = kzalloc(ls->ls_lvblen, ls->ls_allocation);
+        p = kzalloc(ls->ls_lvblen, GFP_NOFS);
         return p;
 }
 
@@ -57,7 +57,7 @@ struct dlm_rsb *dlm_allocate_rsb(struct dlm_ls *ls, int namelen)
 
         DLM_ASSERT(namelen <= DLM_RESNAME_MAXLEN,);
 
-        r = kzalloc(sizeof(*r) + namelen, ls->ls_allocation);
+        r = kzalloc(sizeof(*r) + namelen, GFP_NOFS);
         return r;
 }
 
@@ -72,7 +72,7 @@ struct dlm_lkb *dlm_allocate_lkb(struct dlm_ls *ls)
 {
         struct dlm_lkb *lkb;
 
-        lkb = kmem_cache_zalloc(lkb_cache, ls->ls_allocation);
+        lkb = kmem_cache_zalloc(lkb_cache, GFP_NOFS);
         return lkb;
 }
 
diff --git a/fs/dlm/netlink.c b/fs/dlm/netlink.c
index 55ea369f43a9..2c6ad518100d 100644
--- a/fs/dlm/netlink.c
+++ b/fs/dlm/netlink.c
@@ -9,6 +9,7 @@
 #include <net/genetlink.h>
 #include <linux/dlm.h>
 #include <linux/dlm_netlink.h>
+#include <linux/gfp.h>
 
 #include "dlm_internal.h"
 
@@ -26,7 +27,7 @@ static int prepare_data(u8 cmd, struct sk_buff **skbp, size_t size)
         struct sk_buff *skb;
         void *data;
 
-        skb = genlmsg_new(size, GFP_KERNEL);
+        skb = genlmsg_new(size, GFP_NOFS);
         if (!skb)
                 return -ENOMEM;
 
diff --git a/fs/dlm/plock.c b/fs/dlm/plock.c
index 16f682e26c07..d45c02db6943 100644
--- a/fs/dlm/plock.c
+++ b/fs/dlm/plock.c
@@ -11,6 +11,7 @@
 #include <linux/poll.h>
 #include <linux/dlm.h>
 #include <linux/dlm_plock.h>
+#include <linux/slab.h>
 
 #include "dlm_internal.h"
 #include "lockspace.h"
@@ -82,7 +83,7 @@ int dlm_posix_lock(dlm_lockspace_t *lockspace, u64 number, struct file *file,
         if (!ls)
                 return -EINVAL;
 
-        xop = kzalloc(sizeof(*xop), GFP_KERNEL);
+        xop = kzalloc(sizeof(*xop), GFP_NOFS);
         if (!xop) {
                 rv = -ENOMEM;
                 goto out;
@@ -143,7 +144,7 @@ out:
 }
 EXPORT_SYMBOL_GPL(dlm_posix_lock);
 
-/* Returns failure iff a succesful lock operation should be canceled */
+/* Returns failure iff a successful lock operation should be canceled */
 static int dlm_plock_callback(struct plock_op *op)
 {
         struct file *file;
@@ -211,7 +212,7 @@ int dlm_posix_unlock(dlm_lockspace_t *lockspace, u64 number, struct file *file,
         if (!ls)
                 return -EINVAL;
 
-        op = kzalloc(sizeof(*op), GFP_KERNEL);
+        op = kzalloc(sizeof(*op), GFP_NOFS);
         if (!op) {
                 rv = -ENOMEM;
                 goto out;
@@ -266,7 +267,7 @@ int dlm_posix_get(dlm_lockspace_t *lockspace, u64 number, struct file *file,
         if (!ls)
                 return -EINVAL;
 
-        op = kzalloc(sizeof(*op), GFP_KERNEL);
+        op = kzalloc(sizeof(*op), GFP_NOFS);
         if (!op) {
                 rv = -ENOMEM;
                 goto out;
diff --git a/fs/dlm/rcom.c b/fs/dlm/rcom.c
index 67522c268c14..3c83a49a48a3 100644
--- a/fs/dlm/rcom.c
+++ b/fs/dlm/rcom.c
@@ -38,7 +38,7 @@ static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len,
         char *mb;
         int mb_len = sizeof(struct dlm_rcom) + len;
 
-        mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, ls->ls_allocation, &mb);
+        mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb);
         if (!mh) {
                 log_print("create_rcom to %d type %d len %d ENOBUFS",
                           to_nodeid, type, len);
diff --git a/fs/dlm/requestqueue.c b/fs/dlm/requestqueue.c
index 7a2307c08911..a44fa22890e1 100644
--- a/fs/dlm/requestqueue.c
+++ b/fs/dlm/requestqueue.c
@@ -35,7 +35,7 @@ void dlm_add_requestqueue(struct dlm_ls *ls, int nodeid, struct dlm_message *ms)
         struct rq_entry *e;
         int length = ms->m_header.h_length - sizeof(struct dlm_message);
 
-        e = kmalloc(sizeof(struct rq_entry) + length, ls->ls_allocation);
+        e = kmalloc(sizeof(struct rq_entry) + length, GFP_NOFS);
         if (!e) {
                 log_print("dlm_add_requestqueue: out of memory len %d", length);
                 return;
diff --git a/fs/dlm/user.c b/fs/dlm/user.c
index ebce994ab0b7..8b6e73c47435 100644
--- a/fs/dlm/user.c
+++ b/fs/dlm/user.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2006-2009 Red Hat, Inc.  All rights reserved.
+ * Copyright (C) 2006-2010 Red Hat, Inc.  All rights reserved.
  *
  * This copyrighted material is made available to anyone wishing to use,
  * modify, copy, or redistribute it subject to the terms and conditions
@@ -17,6 +17,7 @@
 #include <linux/spinlock.h>
 #include <linux/dlm.h>
 #include <linux/dlm_device.h>
+#include <linux/slab.h>
 
 #include "dlm_internal.h"
 #include "lockspace.h"
@@ -173,7 +174,7 @@ static int lkb_is_endoflife(struct dlm_lkb *lkb, int sb_status, int type)
 /* we could possibly check if the cancel of an orphan has resulted in the lkb
    being removed and then remove that lkb from the orphans list and free it */
 
-void dlm_user_add_ast(struct dlm_lkb *lkb, int type, int bastmode)
+void dlm_user_add_ast(struct dlm_lkb *lkb, int type, int mode)
 {
         struct dlm_ls *ls;
         struct dlm_user_args *ua;
@@ -206,8 +207,10 @@ void dlm_user_add_ast(struct dlm_lkb *lkb, int type, int bastmode)
 
         ast_type = lkb->lkb_ast_type;
         lkb->lkb_ast_type |= type;
-        if (bastmode)
-                lkb->lkb_bastmode = bastmode;
+        if (type == AST_BAST)
+                lkb->lkb_bastmode = mode;
+        else
+                lkb->lkb_castmode = mode;
 
         if (!ast_type) {
                 kref_get(&lkb->lkb_ref);
@@ -267,7 +270,7 @@ static int device_user_lock(struct dlm_user_proc *proc,
                 goto out;
         }
 
-        ua = kzalloc(sizeof(struct dlm_user_args), GFP_KERNEL);
+        ua = kzalloc(sizeof(struct dlm_user_args), GFP_NOFS);
         if (!ua)
                 goto out;
         ua->proc = proc;
@@ -307,7 +310,7 @@ static int device_user_unlock(struct dlm_user_proc *proc,
         if (!ls)
                 return -ENOENT;
 
-        ua = kzalloc(sizeof(struct dlm_user_args), GFP_KERNEL);
+        ua = kzalloc(sizeof(struct dlm_user_args), GFP_NOFS);
         if (!ua)
                 goto out;
         ua->proc = proc;
@@ -352,7 +355,7 @@ static int dlm_device_register(struct dlm_ls *ls, char *name)
 
         error = -ENOMEM;
         len = strlen(name) + strlen(name_prefix) + 2;
-        ls->ls_device.name = kzalloc(len, GFP_KERNEL);
+        ls->ls_device.name = kzalloc(len, GFP_NOFS);
         if (!ls->ls_device.name)
                 goto fail;
 
@@ -520,7 +523,7 @@ static ssize_t device_write(struct file *file, const char __user *buf,
 #endif
                 return -EINVAL;
 
-        kbuf = kzalloc(count + 1, GFP_KERNEL);
+        kbuf = kzalloc(count + 1, GFP_NOFS);
         if (!kbuf)
                 return -ENOMEM;
 
@@ -546,7 +549,7 @@ static ssize_t device_write(struct file *file, const char __user *buf,
 
                 /* add 1 after namelen so that the name string is terminated */
                 kbuf = kzalloc(sizeof(struct dlm_write_request) + namelen + 1,
-                               GFP_KERNEL);
+                               GFP_NOFS);
                 if (!kbuf) {
                         kfree(k32buf);
                         return -ENOMEM;
@@ -648,7 +651,7 @@ static int device_open(struct inode *inode, struct file *file)
         if (!ls)
                 return -ENOENT;
 
-        proc = kzalloc(sizeof(struct dlm_user_proc), GFP_KERNEL);
+        proc = kzalloc(sizeof(struct dlm_user_proc), GFP_NOFS);
         if (!proc) {
                 dlm_put_lockspace(ls);
                 return -ENOMEM;
diff --git a/fs/dlm/user.h b/fs/dlm/user.h
index 1c9686492286..f196091dd7ff 100644
--- a/fs/dlm/user.h
+++ b/fs/dlm/user.h
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2006-2008 Red Hat, Inc.  All rights reserved.
+ * Copyright (C) 2006-2010 Red Hat, Inc.  All rights reserved.
  *
  * This copyrighted material is made available to anyone wishing to use,
  * modify, copy, or redistribute it subject to the terms and conditions
@@ -9,7 +9,7 @@
 #ifndef __USER_DOT_H__
 #define __USER_DOT_H__
 
-void dlm_user_add_ast(struct dlm_lkb *lkb, int type, int bastmode);
+void dlm_user_add_ast(struct dlm_lkb *lkb, int type, int mode);
 int dlm_user_init(void);
 void dlm_user_exit(void);
 int dlm_device_deregister(struct dlm_ls *ls);