13 files changed, 264 insertions, 78 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 0df243850818..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>
diff --git a/fs/dlm/debug_fs.c b/fs/dlm/debug_fs.c
index 375a2359b3bf..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,
diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h
index 826d3dc6e0ab..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 */
diff --git a/fs/dlm/lock.c b/fs/dlm/lock.c
index 9c0c1db1e105..031dbe3a15ca 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);
+        }
 }
 
 /*
@@ -730,10 +733,7 @@ static void lkb_add_ordered(struct list_head *new, struct list_head *head,
                 if (lkb->lkb_rqmode < mode)
                         break;
 
-        if (!lkb)
-                list_add_tail(new, head);
-        else
-                __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
+        __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue);
 }
 
 /* add/remove lkb to rsb's grant/convert/wait queue */
@@ -2280,20 +2280,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 +2314,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 +2343,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 +2352,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 +2402,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 +2435,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 +2454,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 +2473,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 +2492,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;
 }
@@ -3191,6 +3240,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 +3276,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 +3320,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 +3364,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);
diff --git a/fs/dlm/lockspace.c b/fs/dlm/lockspace.c
index c010ecfc0d29..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;
diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c
index 52cab160893c..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>
 
diff --git a/fs/dlm/member.c b/fs/dlm/member.c
index 84f70bfb0baf..b12532e553f8 100644
--- a/fs/dlm/member.c
+++ b/fs/dlm/member.c
@@ -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.
          */
diff --git a/fs/dlm/netlink.c b/fs/dlm/netlink.c
index 052095cd592f..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"
 
diff --git a/fs/dlm/plock.c b/fs/dlm/plock.c
index b5f89aef3b29..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"
diff --git a/fs/dlm/user.c b/fs/dlm/user.c
index e73a4bb572aa..b6272853130c 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,12 +207,15 @@ 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);
                 list_add_tail(&lkb->lkb_astqueue, &proc->asts);
+                lkb->lkb_ast_first = type;
                 wake_up_interruptible(&proc->wait);
         }
         if (type == AST_COMP && (ast_type & AST_COMP))
@@ -220,7 +224,6 @@ void dlm_user_add_ast(struct dlm_lkb *lkb, int type, int bastmode)
 
         eol = lkb_is_endoflife(lkb, ua->lksb.sb_status, type);
         if (eol) {
-                lkb->lkb_ast_type &= ~AST_BAST;
                 lkb->lkb_flags |= DLM_IFL_ENDOFLIFE;
         }
 
@@ -703,7 +706,7 @@ static int device_close(struct inode *inode, struct file *file)
 }
 
 static int copy_result_to_user(struct dlm_user_args *ua, int compat, int type,
-                               int bmode, char __user *buf, size_t count)
+                               int mode, char __user *buf, size_t count)
 {
 #ifdef CONFIG_COMPAT
         struct dlm_lock_result32 result32;
@@ -730,7 +733,7 @@ static int copy_result_to_user(struct dlm_user_args *ua, int compat, int type,
         if (type == AST_BAST) {
                 result.user_astaddr = ua->bastaddr;
                 result.user_astparam = ua->bastparam;
-                result.bast_mode = bmode;
+                result.bast_mode = mode;
         } else {
                 result.user_astaddr = ua->castaddr;
                 result.user_astparam = ua->castparam;
@@ -798,7 +801,9 @@ static ssize_t device_read(struct file *file, char __user *buf, size_t count,
         struct dlm_user_proc *proc = file->private_data;
         struct dlm_lkb *lkb;
         DECLARE_WAITQUEUE(wait, current);
-        int error, type=0, bmode=0, removed = 0;
+        int error = 0, removed;
+        int ret_type, ret_mode;
+        int bastmode, castmode, do_bast, do_cast;
 
         if (count == sizeof(struct dlm_device_version)) {
                 error = copy_version_to_user(buf, count);
@@ -817,6 +822,8 @@ static ssize_t device_read(struct file *file, char __user *buf, size_t count,
 #endif
                 return -EINVAL;
 
+ try_another:
+
         /* do we really need this? can a read happen after a close? */
         if (test_bit(DLM_PROC_FLAGS_CLOSING, &proc->flags))
                 return -EINVAL;
@@ -852,13 +859,55 @@ static ssize_t device_read(struct file *file, char __user *buf, size_t count,
 
         lkb = list_entry(proc->asts.next, struct dlm_lkb, lkb_astqueue);
 
-        if (lkb->lkb_ast_type & AST_COMP) {
-                lkb->lkb_ast_type &= ~AST_COMP;
-                type = AST_COMP;
-        } else if (lkb->lkb_ast_type & AST_BAST) {
-                lkb->lkb_ast_type &= ~AST_BAST;
-                type = AST_BAST;
-                bmode = lkb->lkb_bastmode;
+        removed = 0;
+        ret_type = 0;
+        ret_mode = 0;
+        do_bast = lkb->lkb_ast_type & AST_BAST;
+        do_cast = lkb->lkb_ast_type & AST_COMP;
+        bastmode = lkb->lkb_bastmode;
+        castmode = lkb->lkb_castmode;
+
+        /* when both are queued figure out which to do first and
+           switch first so the other goes in the next read */
+
+        if (do_cast && do_bast) {
+                if (lkb->lkb_ast_first == AST_COMP) {
+                        ret_type = AST_COMP;
+                        ret_mode = castmode;
+                        lkb->lkb_ast_type &= ~AST_COMP;
+                        lkb->lkb_ast_first = AST_BAST;
+                } else {
+                        ret_type = AST_BAST;
+                        ret_mode = bastmode;
+                        lkb->lkb_ast_type &= ~AST_BAST;
+                        lkb->lkb_ast_first = AST_COMP;
+                }
+        } else {
+                ret_type = lkb->lkb_ast_first;
+                ret_mode = (ret_type == AST_COMP) ? castmode : bastmode;
+                lkb->lkb_ast_type &= ~ret_type;
+                lkb->lkb_ast_first = 0;
+        }
+
+        /* if we're doing a bast but the bast is unnecessary, then
+           switch to do nothing or do a cast if that was needed next */
+
+        if ((ret_type == AST_BAST) &&
+            dlm_modes_compat(bastmode, lkb->lkb_castmode_done)) {
+                ret_type = 0;
+                ret_mode = 0;
+
+                if (do_cast) {
+                        ret_type = AST_COMP;
+                        ret_mode = castmode;
+                        lkb->lkb_ast_type &= ~AST_COMP;
+                        lkb->lkb_ast_first = 0;
+                }
+        }
+
+        if (lkb->lkb_ast_first != lkb->lkb_ast_type) {
+                log_print("device_read %x ast_first %x ast_type %x",
+                          lkb->lkb_id, lkb->lkb_ast_first, lkb->lkb_ast_type);
         }
 
         if (!lkb->lkb_ast_type) {
@@ -867,15 +916,29 @@ static ssize_t device_read(struct file *file, char __user *buf, size_t count,
         }
         spin_unlock(&proc->asts_spin);
 
-        error = copy_result_to_user(lkb->lkb_ua,
-                                test_bit(DLM_PROC_FLAGS_COMPAT, &proc->flags),
-                                type, bmode, buf, count);
+        if (ret_type) {
+                error = copy_result_to_user(lkb->lkb_ua,
+                                test_bit(DLM_PROC_FLAGS_COMPAT, &proc->flags),
+                                ret_type, ret_mode, buf, count);
+
+                if (ret_type == AST_COMP)
+                        lkb->lkb_castmode_done = castmode;
+                if (ret_type == AST_BAST)
+                        lkb->lkb_bastmode_done = bastmode;
+        }
 
         /* removes reference for the proc->asts lists added by
            dlm_user_add_ast() and may result in the lkb being freed */
+
         if (removed)
                 dlm_put_lkb(lkb);
 
+        /* the bast that was queued was eliminated (see unnecessary above),
+           leaving nothing to return */
+
+        if (!ret_type)
+                goto try_another;
+
         return error;
 }
 
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);