ocfs2: Fix setting of OCFS2_LOCK_BLOCKED during bast
[safe/jmp/linux-2.6] / fs / ocfs2 / dlmglue.c
index 058aa86..0cdf630 100644 (file)
@@ -53,6 +53,7 @@
 #include "super.h"
 #include "uptodate.h"
 #include "quota.h"
+#include "refcounttree.h"
 
 #include "buffer_head_io.h"
 
@@ -92,6 +93,9 @@ struct ocfs2_unblock_ctl {
        enum ocfs2_unblock_action unblock_action;
 };
 
+/* Lockdep class keys */
+struct lock_class_key lockdep_keys[OCFS2_NUM_LOCK_TYPES];
+
 static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
                                        int new_level);
 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres);
@@ -107,6 +111,11 @@ static void ocfs2_dentry_post_unlock(struct ocfs2_super *osb,
 
 static void ocfs2_set_qinfo_lvb(struct ocfs2_lock_res *lockres);
 
+static int ocfs2_check_refcount_downconvert(struct ocfs2_lock_res *lockres,
+                                           int new_level);
+static int ocfs2_refcount_convert_worker(struct ocfs2_lock_res *lockres,
+                                        int blocking);
+
 #define mlog_meta_lvb(__level, __lockres) ocfs2_dump_meta_lvb_info(__level, __PRETTY_FUNCTION__, __LINE__, __lockres)
 
 /* This aids in debugging situations where a bad LVB might be involved. */
@@ -115,8 +124,7 @@ static void ocfs2_dump_meta_lvb_info(u64 level,
                                     unsigned int line,
                                     struct ocfs2_lock_res *lockres)
 {
-       struct ocfs2_meta_lvb *lvb =
-               (struct ocfs2_meta_lvb *)ocfs2_dlm_lvb(&lockres->l_lksb);
+       struct ocfs2_meta_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
 
        mlog(level, "LVB information for %s (called from %s:%u):\n",
             lockres->l_name, function, line);
@@ -245,6 +253,14 @@ static struct ocfs2_lock_res_ops ocfs2_rename_lops = {
        .flags          = 0,
 };
 
+static struct ocfs2_lock_res_ops ocfs2_nfs_sync_lops = {
+       .flags          = 0,
+};
+
+static struct ocfs2_lock_res_ops ocfs2_orphan_scan_lops = {
+       .flags          = LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_USES_LVB,
+};
+
 static struct ocfs2_lock_res_ops ocfs2_dentry_lops = {
        .get_osb        = ocfs2_get_dentry_osb,
        .post_unlock    = ocfs2_dentry_post_unlock,
@@ -268,6 +284,12 @@ static struct ocfs2_lock_res_ops ocfs2_qinfo_lops = {
        .flags          = LOCK_TYPE_REQUIRES_REFRESH | LOCK_TYPE_USES_LVB,
 };
 
+static struct ocfs2_lock_res_ops ocfs2_refcount_block_lops = {
+       .check_downconvert = ocfs2_check_refcount_downconvert,
+       .downconvert_worker = ocfs2_refcount_convert_worker,
+       .flags          = 0,
+};
+
 static inline int ocfs2_is_inode_lock(struct ocfs2_lock_res *lockres)
 {
        return lockres->l_type == OCFS2_LOCK_TYPE_META ||
@@ -296,6 +318,12 @@ static inline struct ocfs2_mem_dqinfo *ocfs2_lock_res_qinfo(struct ocfs2_lock_re
        return (struct ocfs2_mem_dqinfo *)lockres->l_priv;
 }
 
+static inline struct ocfs2_refcount_tree *
+ocfs2_lock_res_refcount_tree(struct ocfs2_lock_res *res)
+{
+       return container_of(res, struct ocfs2_refcount_tree, rf_lockres);
+}
+
 static inline struct ocfs2_super *ocfs2_get_lockres_osb(struct ocfs2_lock_res *lockres)
 {
        if (lockres->l_ops->get_osb)
@@ -310,9 +338,16 @@ static int ocfs2_lock_create(struct ocfs2_super *osb,
                             u32 dlm_flags);
 static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lockres,
                                                     int wanted);
-static void ocfs2_cluster_unlock(struct ocfs2_super *osb,
-                                struct ocfs2_lock_res *lockres,
-                                int level);
+static void __ocfs2_cluster_unlock(struct ocfs2_super *osb,
+                                  struct ocfs2_lock_res *lockres,
+                                  int level, unsigned long caller_ip);
+static inline void ocfs2_cluster_unlock(struct ocfs2_super *osb,
+                                       struct ocfs2_lock_res *lockres,
+                                       int level)
+{
+       __ocfs2_cluster_unlock(osb, lockres, level, _RET_IP_);
+}
+
 static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res *lockres);
 static inline void ocfs2_generic_handle_convert_action(struct ocfs2_lock_res *lockres);
 static inline void ocfs2_generic_handle_attach_action(struct ocfs2_lock_res *lockres);
@@ -321,9 +356,14 @@ static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
                                        struct ocfs2_lock_res *lockres);
 static inline void ocfs2_recover_from_dlm_error(struct ocfs2_lock_res *lockres,
                                                int convert);
-#define ocfs2_log_dlm_error(_func, _err, _lockres) do {                        \
-       mlog(ML_ERROR, "DLM error %d while calling %s on resource %s\n", \
-            _err, _func, _lockres->l_name);                            \
+#define ocfs2_log_dlm_error(_func, _err, _lockres) do {                                        \
+       if ((_lockres)->l_type != OCFS2_LOCK_TYPE_DENTRY)                               \
+               mlog(ML_ERROR, "DLM error %d while calling %s on resource %s\n",        \
+                    _err, _func, _lockres->l_name);                                    \
+       else                                                                            \
+               mlog(ML_ERROR, "DLM error %d while calling %s on resource %.*s%08x\n",  \
+                    _err, _func, OCFS2_DENTRY_LOCK_INO_START - 1, (_lockres)->l_name,  \
+                    (unsigned int)ocfs2_get_dentry_lock_ino(_lockres));                \
 } while (0)
 static int ocfs2_downconvert_thread(void *arg);
 static void ocfs2_downconvert_on_unlock(struct ocfs2_super *osb,
@@ -477,6 +517,13 @@ static void ocfs2_lock_res_init_common(struct ocfs2_super *osb,
        ocfs2_add_lockres_tracking(res, osb->osb_dlm_debug);
 
        ocfs2_init_lock_stats(res);
+#ifdef CONFIG_DEBUG_LOCK_ALLOC
+       if (type != OCFS2_LOCK_TYPE_OPEN)
+               lockdep_init_map(&res->l_lockdep_map, ocfs2_lock_type_strings[type],
+                                &lockdep_keys[type], 0);
+       else
+               res->l_lockdep_map.key = NULL;
+#endif
 }
 
 void ocfs2_lock_res_init_once(struct ocfs2_lock_res *res)
@@ -618,6 +665,26 @@ static void ocfs2_rename_lock_res_init(struct ocfs2_lock_res *res,
                                   &ocfs2_rename_lops, osb);
 }
 
+static void ocfs2_nfs_sync_lock_res_init(struct ocfs2_lock_res *res,
+                                        struct ocfs2_super *osb)
+{
+       /* nfs_sync lockres doesn't come from a slab so we call init
+        * once on it manually.  */
+       ocfs2_lock_res_init_once(res);
+       ocfs2_build_lock_name(OCFS2_LOCK_TYPE_NFS_SYNC, 0, 0, res->l_name);
+       ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_NFS_SYNC,
+                                  &ocfs2_nfs_sync_lops, osb);
+}
+
+static void ocfs2_orphan_scan_lock_res_init(struct ocfs2_lock_res *res,
+                                           struct ocfs2_super *osb)
+{
+       ocfs2_lock_res_init_once(res);
+       ocfs2_build_lock_name(OCFS2_LOCK_TYPE_ORPHAN_SCAN, 0, 0, res->l_name);
+       ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_ORPHAN_SCAN,
+                                  &ocfs2_orphan_scan_lops, osb);
+}
+
 void ocfs2_file_lock_res_init(struct ocfs2_lock_res *lockres,
                              struct ocfs2_file_private *fp)
 {
@@ -644,6 +711,17 @@ void ocfs2_qinfo_lock_res_init(struct ocfs2_lock_res *lockres,
                                   info);
 }
 
+void ocfs2_refcount_lock_res_init(struct ocfs2_lock_res *lockres,
+                                 struct ocfs2_super *osb, u64 ref_blkno,
+                                 unsigned int generation)
+{
+       ocfs2_lock_res_init_once(lockres);
+       ocfs2_build_lock_name(OCFS2_LOCK_TYPE_REFCOUNT, ref_blkno,
+                             generation, lockres->l_name);
+       ocfs2_lock_res_init_common(osb, lockres, OCFS2_LOCK_TYPE_REFCOUNT,
+                                  &ocfs2_refcount_block_lops, osb);
+}
+
 void ocfs2_lock_res_free(struct ocfs2_lock_res *res)
 {
        mlog_entry_void();
@@ -829,8 +907,6 @@ static int ocfs2_generic_handle_bast(struct ocfs2_lock_res *lockres,
 
        assert_spin_locked(&lockres->l_lock);
 
-       lockres_or_flags(lockres, OCFS2_LOCK_BLOCKED);
-
        if (level > lockres->l_blocking) {
                /* only schedule a downconvert if we haven't already scheduled
                 * one that goes low enough to satisfy the level we're
@@ -843,6 +919,9 @@ static int ocfs2_generic_handle_bast(struct ocfs2_lock_res *lockres,
                lockres->l_blocking = level;
        }
 
+       if (needs_downconvert)
+               lockres_or_flags(lockres, OCFS2_LOCK_BLOCKED);
+
        mlog_exit(needs_downconvert);
        return needs_downconvert;
 }
@@ -1220,11 +1299,13 @@ static int ocfs2_wait_for_mask_interruptible(struct ocfs2_mask_waiter *mw,
        return ret;
 }
 
-static int ocfs2_cluster_lock(struct ocfs2_super *osb,
-                             struct ocfs2_lock_res *lockres,
-                             int level,
-                             u32 lkm_flags,
-                             int arg_flags)
+static int __ocfs2_cluster_lock(struct ocfs2_super *osb,
+                               struct ocfs2_lock_res *lockres,
+                               int level,
+                               u32 lkm_flags,
+                               int arg_flags,
+                               int l_subclass,
+                               unsigned long caller_ip)
 {
        struct ocfs2_mask_waiter mw;
        int wait, catch_signals = !(osb->s_mount_opt & OCFS2_MOUNT_NOINTR);
@@ -1325,7 +1406,7 @@ again:
                        goto out;
                }
 
-               mlog(0, "lock %s, successfull return from ocfs2_dlm_lock\n",
+               mlog(0, "lock %s, successful return from ocfs2_dlm_lock\n",
                     lockres->l_name);
 
                /* At this point we've gone inside the dlm and need to
@@ -1367,13 +1448,37 @@ out:
        }
        ocfs2_update_lock_stats(lockres, level, &mw, ret);
 
+#ifdef CONFIG_DEBUG_LOCK_ALLOC
+       if (!ret && lockres->l_lockdep_map.key != NULL) {
+               if (level == DLM_LOCK_PR)
+                       rwsem_acquire_read(&lockres->l_lockdep_map, l_subclass,
+                               !!(arg_flags & OCFS2_META_LOCK_NOQUEUE),
+                               caller_ip);
+               else
+                       rwsem_acquire(&lockres->l_lockdep_map, l_subclass,
+                               !!(arg_flags & OCFS2_META_LOCK_NOQUEUE),
+                               caller_ip);
+       }
+#endif
        mlog_exit(ret);
        return ret;
 }
 
-static void ocfs2_cluster_unlock(struct ocfs2_super *osb,
-                                struct ocfs2_lock_res *lockres,
-                                int level)
+static inline int ocfs2_cluster_lock(struct ocfs2_super *osb,
+                                    struct ocfs2_lock_res *lockres,
+                                    int level,
+                                    u32 lkm_flags,
+                                    int arg_flags)
+{
+       return __ocfs2_cluster_lock(osb, lockres, level, lkm_flags, arg_flags,
+                                   0, _RET_IP_);
+}
+
+
+static void __ocfs2_cluster_unlock(struct ocfs2_super *osb,
+                                  struct ocfs2_lock_res *lockres,
+                                  int level,
+                                  unsigned long caller_ip)
 {
        unsigned long flags;
 
@@ -1382,6 +1487,10 @@ static void ocfs2_cluster_unlock(struct ocfs2_super *osb,
        ocfs2_dec_holders(lockres, level);
        ocfs2_downconvert_on_unlock(osb, lockres);
        spin_unlock_irqrestore(&lockres->l_lock, flags);
+#ifdef CONFIG_DEBUG_LOCK_ALLOC
+       if (lockres->l_lockdep_map.key != NULL)
+               rwsem_release(&lockres->l_lockdep_map, 1, caller_ip);
+#endif
        mlog_exit_void();
 }
 
@@ -1469,8 +1578,10 @@ int ocfs2_rw_lock(struct inode *inode, int write)
             (unsigned long long)OCFS2_I(inode)->ip_blkno,
             write ? "EXMODE" : "PRMODE");
 
-       if (ocfs2_mount_local(osb))
+       if (ocfs2_mount_local(osb)) {
+               mlog_exit(0);
                return 0;
+       }
 
        lockres = &OCFS2_I(inode)->ip_rw_lockres;
 
@@ -1745,7 +1856,7 @@ int ocfs2_file_lock(struct file *file, int ex, int trylock)
                 * outstanding lock request, so a cancel convert is
                 * required. We intentionally overwrite 'ret' - if the
                 * cancel fails and the lock was granted, it's easier
-                * to just bubble sucess back up to the user.
+                * to just bubble success back up to the user.
                 */
                ret = ocfs2_flock_handle_signal(lockres, level);
        } else if (!ret && (level > lockres->l_level)) {
@@ -1864,7 +1975,7 @@ static void __ocfs2_stuff_meta_lvb(struct inode *inode)
 
        mlog_entry_void();
 
-       lvb = (struct ocfs2_meta_lvb *)ocfs2_dlm_lvb(&lockres->l_lksb);
+       lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
 
        /*
         * Invalidate the LVB of a deleted inode - this way other
@@ -1916,7 +2027,7 @@ static void ocfs2_refresh_inode_from_lvb(struct inode *inode)
 
        mlog_meta_lvb(0, lockres);
 
-       lvb = (struct ocfs2_meta_lvb *)ocfs2_dlm_lvb(&lockres->l_lksb);
+       lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
 
        /* We're safe here without the lockres lock... */
        spin_lock(&oi->ip_lock);
@@ -1951,10 +2062,10 @@ static void ocfs2_refresh_inode_from_lvb(struct inode *inode)
 static inline int ocfs2_meta_lvb_is_trustable(struct inode *inode,
                                              struct ocfs2_lock_res *lockres)
 {
-       struct ocfs2_meta_lvb *lvb =
-               (struct ocfs2_meta_lvb *)ocfs2_dlm_lvb(&lockres->l_lksb);
+       struct ocfs2_meta_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
 
-       if (lvb->lvb_version == OCFS2_LVB_VERSION
+       if (ocfs2_dlm_lvb_valid(&lockres->l_lksb)
+           && lvb->lvb_version == OCFS2_LVB_VERSION
            && be32_to_cpu(lvb->lvb_igeneration) == inode->i_generation)
                return 1;
        return 0;
@@ -2048,7 +2159,7 @@ static int ocfs2_inode_lock_update(struct inode *inode,
 
        /* This will discard any caching information we might have had
         * for the inode metadata. */
-       ocfs2_metadata_cache_purge(inode);
+       ocfs2_metadata_cache_purge(INODE_CACHE(inode));
 
        ocfs2_extent_map_trunc(inode, 0);
 
@@ -2127,10 +2238,11 @@ static int ocfs2_assign_bh(struct inode *inode,
  * returns < 0 error if the callback will never be called, otherwise
  * the result of the lock will be communicated via the callback.
  */
-int ocfs2_inode_lock_full(struct inode *inode,
-                        struct buffer_head **ret_bh,
-                        int ex,
-                        int arg_flags)
+int ocfs2_inode_lock_full_nested(struct inode *inode,
+                                struct buffer_head **ret_bh,
+                                int ex,
+                                int arg_flags,
+                                int subclass)
 {
        int status, level, acquired;
        u32 dlm_flags;
@@ -2168,7 +2280,8 @@ int ocfs2_inode_lock_full(struct inode *inode,
        if (arg_flags & OCFS2_META_LOCK_NOQUEUE)
                dlm_flags |= DLM_LKF_NOQUEUE;
 
-       status = ocfs2_cluster_lock(osb, lockres, level, dlm_flags, arg_flags);
+       status = __ocfs2_cluster_lock(osb, lockres, level, dlm_flags,
+                                     arg_flags, subclass, _RET_IP_);
        if (status < 0) {
                if (status != -EAGAIN && status != -EIOCBRETRY)
                        mlog_errno(status);
@@ -2334,6 +2447,47 @@ void ocfs2_inode_unlock(struct inode *inode,
        mlog_exit_void();
 }
 
+int ocfs2_orphan_scan_lock(struct ocfs2_super *osb, u32 *seqno)
+{
+       struct ocfs2_lock_res *lockres;
+       struct ocfs2_orphan_scan_lvb *lvb;
+       int status = 0;
+
+       if (ocfs2_is_hard_readonly(osb))
+               return -EROFS;
+
+       if (ocfs2_mount_local(osb))
+               return 0;
+
+       lockres = &osb->osb_orphan_scan.os_lockres;
+       status = ocfs2_cluster_lock(osb, lockres, DLM_LOCK_EX, 0, 0);
+       if (status < 0)
+               return status;
+
+       lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
+       if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
+           lvb->lvb_version == OCFS2_ORPHAN_LVB_VERSION)
+               *seqno = be32_to_cpu(lvb->lvb_os_seqno);
+       else
+               *seqno = osb->osb_orphan_scan.os_seqno + 1;
+
+       return status;
+}
+
+void ocfs2_orphan_scan_unlock(struct ocfs2_super *osb, u32 seqno)
+{
+       struct ocfs2_lock_res *lockres;
+       struct ocfs2_orphan_scan_lvb *lvb;
+
+       if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb)) {
+               lockres = &osb->osb_orphan_scan.os_lockres;
+               lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
+               lvb->lvb_version = OCFS2_ORPHAN_LVB_VERSION;
+               lvb->lvb_os_seqno = cpu_to_be32(seqno);
+               ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
+       }
+}
+
 int ocfs2_super_lock(struct ocfs2_super *osb,
                     int ex)
 {
@@ -2414,6 +2568,34 @@ void ocfs2_rename_unlock(struct ocfs2_super *osb)
                ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
 }
 
+int ocfs2_nfs_sync_lock(struct ocfs2_super *osb, int ex)
+{
+       int status;
+       struct ocfs2_lock_res *lockres = &osb->osb_nfs_sync_lockres;
+
+       if (ocfs2_is_hard_readonly(osb))
+               return -EROFS;
+
+       if (ocfs2_mount_local(osb))
+               return 0;
+
+       status = ocfs2_cluster_lock(osb, lockres, ex ? LKM_EXMODE : LKM_PRMODE,
+                                   0, 0);
+       if (status < 0)
+               mlog(ML_ERROR, "lock on nfs sync lock failed %d\n", status);
+
+       return status;
+}
+
+void ocfs2_nfs_sync_unlock(struct ocfs2_super *osb, int ex)
+{
+       struct ocfs2_lock_res *lockres = &osb->osb_nfs_sync_lockres;
+
+       if (!ocfs2_mount_local(osb))
+               ocfs2_cluster_unlock(osb, lockres,
+                                    ex ? LKM_EXMODE : LKM_PRMODE);
+}
+
 int ocfs2_dentry_lock(struct dentry *dentry, int ex)
 {
        int ret;
@@ -2795,6 +2977,8 @@ int ocfs2_dlm_init(struct ocfs2_super *osb)
 local:
        ocfs2_super_lock_res_init(&osb->osb_super_lockres, osb);
        ocfs2_rename_lock_res_init(&osb->osb_rename_lockres, osb);
+       ocfs2_nfs_sync_lock_res_init(&osb->osb_nfs_sync_lockres, osb);
+       ocfs2_orphan_scan_lock_res_init(&osb->osb_orphan_scan.os_lockres, osb);
 
        osb->cconn = conn;
 
@@ -2830,6 +3014,8 @@ void ocfs2_dlm_shutdown(struct ocfs2_super *osb,
 
        ocfs2_lock_res_free(&osb->osb_super_lockres);
        ocfs2_lock_res_free(&osb->osb_rename_lockres);
+       ocfs2_lock_res_free(&osb->osb_nfs_sync_lockres);
+       ocfs2_lock_res_free(&osb->osb_orphan_scan.os_lockres);
 
        ocfs2_cluster_disconnect(osb->cconn, hangup_pending);
        osb->cconn = NULL;
@@ -2855,6 +3041,7 @@ static void ocfs2_unlock_ast(void *opaque, int error)
                     "unlock_action %d\n", error, lockres->l_name,
                     lockres->l_unlock_action);
                spin_unlock_irqrestore(&lockres->l_lock, flags);
+               mlog_exit_void();
                return;
        }
 
@@ -2862,6 +3049,10 @@ static void ocfs2_unlock_ast(void *opaque, int error)
        case OCFS2_UNLOCK_CANCEL_CONVERT:
                mlog(0, "Cancel convert success for %s\n", lockres->l_name);
                lockres->l_action = OCFS2_AST_INVALID;
+               /* Downconvert thread may have requeued this lock, we
+                * need to wake it. */
+               if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
+                       ocfs2_wake_downconvert_thread(ocfs2_get_lockres_osb(lockres));
                break;
        case OCFS2_UNLOCK_DROP_LOCK:
                lockres->l_level = DLM_LOCK_IV;
@@ -2953,7 +3144,7 @@ static int ocfs2_drop_lock(struct ocfs2_super *osb,
                ocfs2_dlm_dump_lksb(&lockres->l_lksb);
                BUG();
        }
-       mlog(0, "lock %s, successfull return from ocfs2_dlm_unlock\n",
+       mlog(0, "lock %s, successful return from ocfs2_dlm_unlock\n",
             lockres->l_name);
 
        ocfs2_wait_on_busy_lock(lockres);
@@ -2965,7 +3156,7 @@ out:
 /* Mark the lockres as being dropped. It will no longer be
  * queued if blocking, but we still may have to wait on it
  * being dequeued from the downconvert thread before we can consider
- * it safe to drop. 
+ * it safe to drop.
  *
  * You can *not* attempt to call cluster_lock on this lockres anymore. */
 void ocfs2_mark_lockres_freeing(struct ocfs2_lock_res *lockres)
@@ -3008,6 +3199,8 @@ static void ocfs2_drop_osb_locks(struct ocfs2_super *osb)
 {
        ocfs2_simple_drop_lockres(osb, &osb->osb_super_lockres);
        ocfs2_simple_drop_lockres(osb, &osb->osb_rename_lockres);
+       ocfs2_simple_drop_lockres(osb, &osb->osb_nfs_sync_lockres);
+       ocfs2_simple_drop_lockres(osb, &osb->osb_orphan_scan.os_lockres);
 }
 
 int ocfs2_drop_inode_locks(struct inode *inode)
@@ -3335,11 +3528,11 @@ out:
        return UNBLOCK_CONTINUE;
 }
 
-static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
-                                       int new_level)
+static int ocfs2_ci_checkpointed(struct ocfs2_caching_info *ci,
+                                struct ocfs2_lock_res *lockres,
+                                int new_level)
 {
-       struct inode *inode = ocfs2_lock_res_inode(lockres);
-       int checkpointed = ocfs2_inode_fully_checkpointed(inode);
+       int checkpointed = ocfs2_ci_fully_checkpointed(ci);
 
        BUG_ON(new_level != DLM_LOCK_NL && new_level != DLM_LOCK_PR);
        BUG_ON(lockres->l_level != DLM_LOCK_EX && !checkpointed);
@@ -3347,10 +3540,18 @@ static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
        if (checkpointed)
                return 1;
 
-       ocfs2_start_checkpoint(OCFS2_SB(inode->i_sb));
+       ocfs2_start_checkpoint(OCFS2_SB(ocfs2_metadata_cache_get_super(ci)));
        return 0;
 }
 
+static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
+                                       int new_level)
+{
+       struct inode *inode = ocfs2_lock_res_inode(lockres);
+
+       return ocfs2_ci_checkpointed(INODE_CACHE(inode), lockres, new_level);
+}
+
 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres)
 {
        struct inode *inode = ocfs2_lock_res_inode(lockres);
@@ -3480,6 +3681,26 @@ static int ocfs2_dentry_convert_worker(struct ocfs2_lock_res *lockres,
        return UNBLOCK_CONTINUE_POST;
 }
 
+static int ocfs2_check_refcount_downconvert(struct ocfs2_lock_res *lockres,
+                                           int new_level)
+{
+       struct ocfs2_refcount_tree *tree =
+                               ocfs2_lock_res_refcount_tree(lockres);
+
+       return ocfs2_ci_checkpointed(&tree->rf_ci, lockres, new_level);
+}
+
+static int ocfs2_refcount_convert_worker(struct ocfs2_lock_res *lockres,
+                                        int blocking)
+{
+       struct ocfs2_refcount_tree *tree =
+                               ocfs2_lock_res_refcount_tree(lockres);
+
+       ocfs2_metadata_cache_purge(&tree->rf_ci);
+
+       return UNBLOCK_CONTINUE;
+}
+
 static void ocfs2_set_qinfo_lvb(struct ocfs2_lock_res *lockres)
 {
        struct ocfs2_qinfo_lvb *lvb;
@@ -3489,7 +3710,7 @@ static void ocfs2_set_qinfo_lvb(struct ocfs2_lock_res *lockres)
 
        mlog_entry_void();
 
-       lvb = (struct ocfs2_qinfo_lvb *)ocfs2_dlm_lvb(&lockres->l_lksb);
+       lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
        lvb->lvb_version = OCFS2_QINFO_LVB_VERSION;
        lvb->lvb_bgrace = cpu_to_be32(info->dqi_bgrace);
        lvb->lvb_igrace = cpu_to_be32(info->dqi_igrace);
@@ -3519,11 +3740,12 @@ static int ocfs2_refresh_qinfo(struct ocfs2_mem_dqinfo *oinfo)
                                            oinfo->dqi_gi.dqi_type);
        struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
        struct ocfs2_qinfo_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
-       struct buffer_head *bh;
+       struct buffer_head *bh = NULL;
        struct ocfs2_global_disk_dqinfo *gdinfo;
        int status = 0;
 
-       if (lvb->lvb_version == OCFS2_QINFO_LVB_VERSION) {
+       if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
+           lvb->lvb_version == OCFS2_QINFO_LVB_VERSION) {
                info->dqi_bgrace = be32_to_cpu(lvb->lvb_bgrace);
                info->dqi_igrace = be32_to_cpu(lvb->lvb_igrace);
                oinfo->dqi_syncms = be32_to_cpu(lvb->lvb_syncms);
@@ -3532,8 +3754,8 @@ static int ocfs2_refresh_qinfo(struct ocfs2_mem_dqinfo *oinfo)
                oinfo->dqi_gi.dqi_free_entry =
                                        be32_to_cpu(lvb->lvb_free_entry);
        } else {
-               bh = ocfs2_read_quota_block(oinfo->dqi_gqinode, 0, &status);
-               if (!bh) {
+               status = ocfs2_read_quota_block(oinfo->dqi_gqinode, 0, &bh);
+               if (status) {
                        mlog_errno(status);
                        goto bail;
                }
@@ -3591,6 +3813,37 @@ bail:
        return status;
 }
 
+int ocfs2_refcount_lock(struct ocfs2_refcount_tree *ref_tree, int ex)
+{
+       int status;
+       int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
+       struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
+       struct ocfs2_super *osb = lockres->l_priv;
+
+
+       if (ocfs2_is_hard_readonly(osb))
+               return -EROFS;
+
+       if (ocfs2_mount_local(osb))
+               return 0;
+
+       status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
+       if (status < 0)
+               mlog_errno(status);
+
+       return status;
+}
+
+void ocfs2_refcount_unlock(struct ocfs2_refcount_tree *ref_tree, int ex)
+{
+       int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
+       struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
+       struct ocfs2_super *osb = lockres->l_priv;
+
+       if (!ocfs2_mount_local(osb))
+               ocfs2_cluster_unlock(osb, lockres, level);
+}
+
 /*
  * This is the filesystem locking protocol.  It provides the lock handling
  * hooks for the underlying DLM.  It has a maximum version number.