1 // SPDX-License-Identifier: GPL-2.0-or-later
5 * Code which implements an OCFS2 specific interface to our DLM.
7 * Copyright (C) 2003, 2004 Oracle. All rights reserved.
10 #include <linux/types.h>
11 #include <linux/slab.h>
12 #include <linux/highmem.h>
14 #include <linux/kthread.h>
15 #include <linux/pagemap.h>
16 #include <linux/debugfs.h>
17 #include <linux/seq_file.h>
18 #include <linux/time.h>
19 #include <linux/delay.h>
20 #include <linux/quotaops.h>
21 #include <linux/sched/signal.h>
23 #define MLOG_MASK_PREFIX ML_DLM_GLUE
24 #include <cluster/masklog.h>
27 #include "ocfs2_lockingver.h"
32 #include "extent_map.h"
34 #include "heartbeat.h"
37 #include "stackglue.h"
42 #include "refcounttree.h"
45 #include "buffer_head_io.h"
47 struct ocfs2_mask_waiter {
48 struct list_head mw_item;
50 struct completion mw_complete;
51 unsigned long mw_mask;
52 unsigned long mw_goal;
53 #ifdef CONFIG_OCFS2_FS_STATS
54 ktime_t mw_lock_start;
58 static struct ocfs2_super *ocfs2_get_dentry_osb(struct ocfs2_lock_res *lockres);
59 static struct ocfs2_super *ocfs2_get_inode_osb(struct ocfs2_lock_res *lockres);
60 static struct ocfs2_super *ocfs2_get_file_osb(struct ocfs2_lock_res *lockres);
61 static struct ocfs2_super *ocfs2_get_qinfo_osb(struct ocfs2_lock_res *lockres);
64 * Return value from ->downconvert_worker functions.
66 * These control the precise actions of ocfs2_unblock_lock()
67 * and ocfs2_process_blocked_lock()
70 enum ocfs2_unblock_action {
71 UNBLOCK_CONTINUE = 0, /* Continue downconvert */
72 UNBLOCK_CONTINUE_POST = 1, /* Continue downconvert, fire
73 * ->post_unlock callback */
74 UNBLOCK_STOP_POST = 2, /* Do not downconvert, fire
75 * ->post_unlock() callback. */
78 struct ocfs2_unblock_ctl {
80 enum ocfs2_unblock_action unblock_action;
83 /* Lockdep class keys */
84 #ifdef CONFIG_DEBUG_LOCK_ALLOC
85 static struct lock_class_key lockdep_keys[OCFS2_NUM_LOCK_TYPES];
88 static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
90 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres);
92 static int ocfs2_data_convert_worker(struct ocfs2_lock_res *lockres,
95 static int ocfs2_dentry_convert_worker(struct ocfs2_lock_res *lockres,
98 static void ocfs2_dentry_post_unlock(struct ocfs2_super *osb,
99 struct ocfs2_lock_res *lockres);
101 static void ocfs2_set_qinfo_lvb(struct ocfs2_lock_res *lockres);
103 static int ocfs2_check_refcount_downconvert(struct ocfs2_lock_res *lockres,
105 static int ocfs2_refcount_convert_worker(struct ocfs2_lock_res *lockres,
108 #define mlog_meta_lvb(__level, __lockres) ocfs2_dump_meta_lvb_info(__level, __PRETTY_FUNCTION__, __LINE__, __lockres)
110 /* This aids in debugging situations where a bad LVB might be involved. */
111 static void ocfs2_dump_meta_lvb_info(u64 level,
112 const char *function,
114 struct ocfs2_lock_res *lockres)
116 struct ocfs2_meta_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
118 mlog(level, "LVB information for %s (called from %s:%u):\n",
119 lockres->l_name, function, line);
120 mlog(level, "version: %u, clusters: %u, generation: 0x%x\n",
121 lvb->lvb_version, be32_to_cpu(lvb->lvb_iclusters),
122 be32_to_cpu(lvb->lvb_igeneration));
123 mlog(level, "size: %llu, uid %u, gid %u, mode 0x%x\n",
124 (unsigned long long)be64_to_cpu(lvb->lvb_isize),
125 be32_to_cpu(lvb->lvb_iuid), be32_to_cpu(lvb->lvb_igid),
126 be16_to_cpu(lvb->lvb_imode));
127 mlog(level, "nlink %u, atime_packed 0x%llx, ctime_packed 0x%llx, "
128 "mtime_packed 0x%llx iattr 0x%x\n", be16_to_cpu(lvb->lvb_inlink),
129 (long long)be64_to_cpu(lvb->lvb_iatime_packed),
130 (long long)be64_to_cpu(lvb->lvb_ictime_packed),
131 (long long)be64_to_cpu(lvb->lvb_imtime_packed),
132 be32_to_cpu(lvb->lvb_iattr));
137 * OCFS2 Lock Resource Operations
139 * These fine tune the behavior of the generic dlmglue locking infrastructure.
141 * The most basic of lock types can point ->l_priv to their respective
142 * struct ocfs2_super and allow the default actions to manage things.
144 * Right now, each lock type also needs to implement an init function,
145 * and trivial lock/unlock wrappers. ocfs2_simple_drop_lockres()
146 * should be called when the lock is no longer needed (i.e., object
149 struct ocfs2_lock_res_ops {
151 * Translate an ocfs2_lock_res * into an ocfs2_super *. Define
152 * this callback if ->l_priv is not an ocfs2_super pointer
154 struct ocfs2_super * (*get_osb)(struct ocfs2_lock_res *);
157 * Optionally called in the downconvert thread after a
158 * successful downconvert. The lockres will not be referenced
159 * after this callback is called, so it is safe to free
162 * The exact semantics of when this is called are controlled
163 * by ->downconvert_worker()
165 void (*post_unlock)(struct ocfs2_super *, struct ocfs2_lock_res *);
168 * Allow a lock type to add checks to determine whether it is
169 * safe to downconvert a lock. Return 0 to re-queue the
170 * downconvert at a later time, nonzero to continue.
172 * For most locks, the default checks that there are no
173 * incompatible holders are sufficient.
175 * Called with the lockres spinlock held.
177 int (*check_downconvert)(struct ocfs2_lock_res *, int);
180 * Allows a lock type to populate the lock value block. This
181 * is called on downconvert, and when we drop a lock.
183 * Locks that want to use this should set LOCK_TYPE_USES_LVB
184 * in the flags field.
186 * Called with the lockres spinlock held.
188 void (*set_lvb)(struct ocfs2_lock_res *);
191 * Called from the downconvert thread when it is determined
192 * that a lock will be downconverted. This is called without
193 * any locks held so the function can do work that might
194 * schedule (syncing out data, etc).
196 * This should return any one of the ocfs2_unblock_action
197 * values, depending on what it wants the thread to do.
199 int (*downconvert_worker)(struct ocfs2_lock_res *, int);
202 * LOCK_TYPE_* flags which describe the specific requirements
203 * of a lock type. Descriptions of each individual flag follow.
209 * Some locks want to "refresh" potentially stale data when a
210 * meaningful (PRMODE or EXMODE) lock level is first obtained. If this
211 * flag is set, the OCFS2_LOCK_NEEDS_REFRESH flag will be set on the
212 * individual lockres l_flags member from the ast function. It is
213 * expected that the locking wrapper will clear the
214 * OCFS2_LOCK_NEEDS_REFRESH flag when done.
216 #define LOCK_TYPE_REQUIRES_REFRESH 0x1
219 * Indicate that a lock type makes use of the lock value block. The
220 * ->set_lvb lock type callback must be defined.
222 #define LOCK_TYPE_USES_LVB 0x2
224 static struct ocfs2_lock_res_ops ocfs2_inode_rw_lops = {
225 .get_osb = ocfs2_get_inode_osb,
229 static struct ocfs2_lock_res_ops ocfs2_inode_inode_lops = {
230 .get_osb = ocfs2_get_inode_osb,
231 .check_downconvert = ocfs2_check_meta_downconvert,
232 .set_lvb = ocfs2_set_meta_lvb,
233 .downconvert_worker = ocfs2_data_convert_worker,
234 .flags = LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_USES_LVB,
237 static struct ocfs2_lock_res_ops ocfs2_super_lops = {
238 .flags = LOCK_TYPE_REQUIRES_REFRESH,
241 static struct ocfs2_lock_res_ops ocfs2_rename_lops = {
245 static struct ocfs2_lock_res_ops ocfs2_nfs_sync_lops = {
249 static struct ocfs2_lock_res_ops ocfs2_trim_fs_lops = {
250 .flags = LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_USES_LVB,
253 static struct ocfs2_lock_res_ops ocfs2_orphan_scan_lops = {
254 .flags = LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_USES_LVB,
257 static struct ocfs2_lock_res_ops ocfs2_dentry_lops = {
258 .get_osb = ocfs2_get_dentry_osb,
259 .post_unlock = ocfs2_dentry_post_unlock,
260 .downconvert_worker = ocfs2_dentry_convert_worker,
264 static struct ocfs2_lock_res_ops ocfs2_inode_open_lops = {
265 .get_osb = ocfs2_get_inode_osb,
269 static struct ocfs2_lock_res_ops ocfs2_flock_lops = {
270 .get_osb = ocfs2_get_file_osb,
274 static struct ocfs2_lock_res_ops ocfs2_qinfo_lops = {
275 .set_lvb = ocfs2_set_qinfo_lvb,
276 .get_osb = ocfs2_get_qinfo_osb,
277 .flags = LOCK_TYPE_REQUIRES_REFRESH | LOCK_TYPE_USES_LVB,
280 static struct ocfs2_lock_res_ops ocfs2_refcount_block_lops = {
281 .check_downconvert = ocfs2_check_refcount_downconvert,
282 .downconvert_worker = ocfs2_refcount_convert_worker,
286 static inline int ocfs2_is_inode_lock(struct ocfs2_lock_res *lockres)
288 return lockres->l_type == OCFS2_LOCK_TYPE_META ||
289 lockres->l_type == OCFS2_LOCK_TYPE_RW ||
290 lockres->l_type == OCFS2_LOCK_TYPE_OPEN;
293 static inline struct ocfs2_lock_res *ocfs2_lksb_to_lock_res(struct ocfs2_dlm_lksb *lksb)
295 return container_of(lksb, struct ocfs2_lock_res, l_lksb);
298 static inline struct inode *ocfs2_lock_res_inode(struct ocfs2_lock_res *lockres)
300 BUG_ON(!ocfs2_is_inode_lock(lockres));
302 return (struct inode *) lockres->l_priv;
305 static inline struct ocfs2_dentry_lock *ocfs2_lock_res_dl(struct ocfs2_lock_res *lockres)
307 BUG_ON(lockres->l_type != OCFS2_LOCK_TYPE_DENTRY);
309 return (struct ocfs2_dentry_lock *)lockres->l_priv;
312 static inline struct ocfs2_mem_dqinfo *ocfs2_lock_res_qinfo(struct ocfs2_lock_res *lockres)
314 BUG_ON(lockres->l_type != OCFS2_LOCK_TYPE_QINFO);
316 return (struct ocfs2_mem_dqinfo *)lockres->l_priv;
319 static inline struct ocfs2_refcount_tree *
320 ocfs2_lock_res_refcount_tree(struct ocfs2_lock_res *res)
322 return container_of(res, struct ocfs2_refcount_tree, rf_lockres);
325 static inline struct ocfs2_super *ocfs2_get_lockres_osb(struct ocfs2_lock_res *lockres)
327 if (lockres->l_ops->get_osb)
328 return lockres->l_ops->get_osb(lockres);
330 return (struct ocfs2_super *)lockres->l_priv;
333 static int ocfs2_lock_create(struct ocfs2_super *osb,
334 struct ocfs2_lock_res *lockres,
337 static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lockres,
339 static void __ocfs2_cluster_unlock(struct ocfs2_super *osb,
340 struct ocfs2_lock_res *lockres,
341 int level, unsigned long caller_ip);
342 static inline void ocfs2_cluster_unlock(struct ocfs2_super *osb,
343 struct ocfs2_lock_res *lockres,
346 __ocfs2_cluster_unlock(osb, lockres, level, _RET_IP_);
349 static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res *lockres);
350 static inline void ocfs2_generic_handle_convert_action(struct ocfs2_lock_res *lockres);
351 static inline void ocfs2_generic_handle_attach_action(struct ocfs2_lock_res *lockres);
352 static int ocfs2_generic_handle_bast(struct ocfs2_lock_res *lockres, int level);
353 static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
354 struct ocfs2_lock_res *lockres);
355 static inline void ocfs2_recover_from_dlm_error(struct ocfs2_lock_res *lockres,
357 #define ocfs2_log_dlm_error(_func, _err, _lockres) do { \
358 if ((_lockres)->l_type != OCFS2_LOCK_TYPE_DENTRY) \
359 mlog(ML_ERROR, "DLM error %d while calling %s on resource %s\n", \
360 _err, _func, _lockres->l_name); \
362 mlog(ML_ERROR, "DLM error %d while calling %s on resource %.*s%08x\n", \
363 _err, _func, OCFS2_DENTRY_LOCK_INO_START - 1, (_lockres)->l_name, \
364 (unsigned int)ocfs2_get_dentry_lock_ino(_lockres)); \
366 static int ocfs2_downconvert_thread(void *arg);
367 static void ocfs2_downconvert_on_unlock(struct ocfs2_super *osb,
368 struct ocfs2_lock_res *lockres);
369 static int ocfs2_inode_lock_update(struct inode *inode,
370 struct buffer_head **bh);
371 static void ocfs2_drop_osb_locks(struct ocfs2_super *osb);
372 static inline int ocfs2_highest_compat_lock_level(int level);
373 static unsigned int ocfs2_prepare_downconvert(struct ocfs2_lock_res *lockres,
375 static int ocfs2_downconvert_lock(struct ocfs2_super *osb,
376 struct ocfs2_lock_res *lockres,
379 unsigned int generation);
380 static int ocfs2_prepare_cancel_convert(struct ocfs2_super *osb,
381 struct ocfs2_lock_res *lockres);
382 static int ocfs2_cancel_convert(struct ocfs2_super *osb,
383 struct ocfs2_lock_res *lockres);
386 static void ocfs2_build_lock_name(enum ocfs2_lock_type type,
393 BUG_ON(type >= OCFS2_NUM_LOCK_TYPES);
395 len = snprintf(name, OCFS2_LOCK_ID_MAX_LEN, "%c%s%016llx%08x",
396 ocfs2_lock_type_char(type), OCFS2_LOCK_ID_PAD,
397 (long long)blkno, generation);
399 BUG_ON(len != (OCFS2_LOCK_ID_MAX_LEN - 1));
401 mlog(0, "built lock resource with name: %s\n", name);
404 static DEFINE_SPINLOCK(ocfs2_dlm_tracking_lock);
406 static void ocfs2_add_lockres_tracking(struct ocfs2_lock_res *res,
407 struct ocfs2_dlm_debug *dlm_debug)
409 mlog(0, "Add tracking for lockres %s\n", res->l_name);
411 spin_lock(&ocfs2_dlm_tracking_lock);
412 list_add(&res->l_debug_list, &dlm_debug->d_lockres_tracking);
413 spin_unlock(&ocfs2_dlm_tracking_lock);
416 static void ocfs2_remove_lockres_tracking(struct ocfs2_lock_res *res)
418 spin_lock(&ocfs2_dlm_tracking_lock);
419 if (!list_empty(&res->l_debug_list))
420 list_del_init(&res->l_debug_list);
421 spin_unlock(&ocfs2_dlm_tracking_lock);
424 #ifdef CONFIG_OCFS2_FS_STATS
425 static void ocfs2_init_lock_stats(struct ocfs2_lock_res *res)
427 res->l_lock_refresh = 0;
428 res->l_lock_wait = 0;
429 memset(&res->l_lock_prmode, 0, sizeof(struct ocfs2_lock_stats));
430 memset(&res->l_lock_exmode, 0, sizeof(struct ocfs2_lock_stats));
433 static void ocfs2_update_lock_stats(struct ocfs2_lock_res *res, int level,
434 struct ocfs2_mask_waiter *mw, int ret)
438 struct ocfs2_lock_stats *stats;
440 if (level == LKM_PRMODE)
441 stats = &res->l_lock_prmode;
442 else if (level == LKM_EXMODE)
443 stats = &res->l_lock_exmode;
447 kt = ktime_sub(ktime_get(), mw->mw_lock_start);
448 usec = ktime_to_us(kt);
451 stats->ls_total += ktime_to_ns(kt);
453 if (unlikely(stats->ls_gets == 0)) {
455 stats->ls_total = ktime_to_ns(kt);
458 if (stats->ls_max < usec)
459 stats->ls_max = usec;
464 stats->ls_last = ktime_to_us(ktime_get_real());
467 static inline void ocfs2_track_lock_refresh(struct ocfs2_lock_res *lockres)
469 lockres->l_lock_refresh++;
472 static inline void ocfs2_track_lock_wait(struct ocfs2_lock_res *lockres)
474 struct ocfs2_mask_waiter *mw;
476 if (list_empty(&lockres->l_mask_waiters)) {
477 lockres->l_lock_wait = 0;
481 mw = list_first_entry(&lockres->l_mask_waiters,
482 struct ocfs2_mask_waiter, mw_item);
483 lockres->l_lock_wait =
484 ktime_to_us(ktime_mono_to_real(mw->mw_lock_start));
487 static inline void ocfs2_init_start_time(struct ocfs2_mask_waiter *mw)
489 mw->mw_lock_start = ktime_get();
492 static inline void ocfs2_init_lock_stats(struct ocfs2_lock_res *res)
495 static inline void ocfs2_update_lock_stats(struct ocfs2_lock_res *res,
496 int level, struct ocfs2_mask_waiter *mw, int ret)
499 static inline void ocfs2_track_lock_refresh(struct ocfs2_lock_res *lockres)
502 static inline void ocfs2_track_lock_wait(struct ocfs2_lock_res *lockres)
505 static inline void ocfs2_init_start_time(struct ocfs2_mask_waiter *mw)
510 static void ocfs2_lock_res_init_common(struct ocfs2_super *osb,
511 struct ocfs2_lock_res *res,
512 enum ocfs2_lock_type type,
513 struct ocfs2_lock_res_ops *ops,
520 res->l_level = DLM_LOCK_IV;
521 res->l_requested = DLM_LOCK_IV;
522 res->l_blocking = DLM_LOCK_IV;
523 res->l_action = OCFS2_AST_INVALID;
524 res->l_unlock_action = OCFS2_UNLOCK_INVALID;
526 res->l_flags = OCFS2_LOCK_INITIALIZED;
528 ocfs2_add_lockres_tracking(res, osb->osb_dlm_debug);
530 ocfs2_init_lock_stats(res);
531 #ifdef CONFIG_DEBUG_LOCK_ALLOC
532 if (type != OCFS2_LOCK_TYPE_OPEN)
533 lockdep_init_map(&res->l_lockdep_map, ocfs2_lock_type_strings[type],
534 &lockdep_keys[type], 0);
536 res->l_lockdep_map.key = NULL;
540 void ocfs2_lock_res_init_once(struct ocfs2_lock_res *res)
542 /* This also clears out the lock status block */
543 memset(res, 0, sizeof(struct ocfs2_lock_res));
544 spin_lock_init(&res->l_lock);
545 init_waitqueue_head(&res->l_event);
546 INIT_LIST_HEAD(&res->l_blocked_list);
547 INIT_LIST_HEAD(&res->l_mask_waiters);
548 INIT_LIST_HEAD(&res->l_holders);
551 void ocfs2_inode_lock_res_init(struct ocfs2_lock_res *res,
552 enum ocfs2_lock_type type,
553 unsigned int generation,
556 struct ocfs2_lock_res_ops *ops;
559 case OCFS2_LOCK_TYPE_RW:
560 ops = &ocfs2_inode_rw_lops;
562 case OCFS2_LOCK_TYPE_META:
563 ops = &ocfs2_inode_inode_lops;
565 case OCFS2_LOCK_TYPE_OPEN:
566 ops = &ocfs2_inode_open_lops;
569 mlog_bug_on_msg(1, "type: %d\n", type);
570 ops = NULL; /* thanks, gcc */
574 ocfs2_build_lock_name(type, OCFS2_I(inode)->ip_blkno,
575 generation, res->l_name);
576 ocfs2_lock_res_init_common(OCFS2_SB(inode->i_sb), res, type, ops, inode);
579 static struct ocfs2_super *ocfs2_get_inode_osb(struct ocfs2_lock_res *lockres)
581 struct inode *inode = ocfs2_lock_res_inode(lockres);
583 return OCFS2_SB(inode->i_sb);
586 static struct ocfs2_super *ocfs2_get_qinfo_osb(struct ocfs2_lock_res *lockres)
588 struct ocfs2_mem_dqinfo *info = lockres->l_priv;
590 return OCFS2_SB(info->dqi_gi.dqi_sb);
593 static struct ocfs2_super *ocfs2_get_file_osb(struct ocfs2_lock_res *lockres)
595 struct ocfs2_file_private *fp = lockres->l_priv;
597 return OCFS2_SB(fp->fp_file->f_mapping->host->i_sb);
600 static __u64 ocfs2_get_dentry_lock_ino(struct ocfs2_lock_res *lockres)
602 __be64 inode_blkno_be;
604 memcpy(&inode_blkno_be, &lockres->l_name[OCFS2_DENTRY_LOCK_INO_START],
607 return be64_to_cpu(inode_blkno_be);
610 static struct ocfs2_super *ocfs2_get_dentry_osb(struct ocfs2_lock_res *lockres)
612 struct ocfs2_dentry_lock *dl = lockres->l_priv;
614 return OCFS2_SB(dl->dl_inode->i_sb);
617 void ocfs2_dentry_lock_res_init(struct ocfs2_dentry_lock *dl,
618 u64 parent, struct inode *inode)
621 u64 inode_blkno = OCFS2_I(inode)->ip_blkno;
622 __be64 inode_blkno_be = cpu_to_be64(inode_blkno);
623 struct ocfs2_lock_res *lockres = &dl->dl_lockres;
625 ocfs2_lock_res_init_once(lockres);
628 * Unfortunately, the standard lock naming scheme won't work
629 * here because we have two 16 byte values to use. Instead,
630 * we'll stuff the inode number as a binary value. We still
631 * want error prints to show something without garbling the
632 * display, so drop a null byte in there before the inode
633 * number. A future version of OCFS2 will likely use all
634 * binary lock names. The stringified names have been a
635 * tremendous aid in debugging, but now that the debugfs
636 * interface exists, we can mangle things there if need be.
638 * NOTE: We also drop the standard "pad" value (the total lock
639 * name size stays the same though - the last part is all
640 * zeros due to the memset in ocfs2_lock_res_init_once()
642 len = snprintf(lockres->l_name, OCFS2_DENTRY_LOCK_INO_START,
644 ocfs2_lock_type_char(OCFS2_LOCK_TYPE_DENTRY),
647 BUG_ON(len != (OCFS2_DENTRY_LOCK_INO_START - 1));
649 memcpy(&lockres->l_name[OCFS2_DENTRY_LOCK_INO_START], &inode_blkno_be,
652 ocfs2_lock_res_init_common(OCFS2_SB(inode->i_sb), lockres,
653 OCFS2_LOCK_TYPE_DENTRY, &ocfs2_dentry_lops,
657 static void ocfs2_super_lock_res_init(struct ocfs2_lock_res *res,
658 struct ocfs2_super *osb)
660 /* Superblock lockres doesn't come from a slab so we call init
661 * once on it manually. */
662 ocfs2_lock_res_init_once(res);
663 ocfs2_build_lock_name(OCFS2_LOCK_TYPE_SUPER, OCFS2_SUPER_BLOCK_BLKNO,
665 ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_SUPER,
666 &ocfs2_super_lops, osb);
669 static void ocfs2_rename_lock_res_init(struct ocfs2_lock_res *res,
670 struct ocfs2_super *osb)
672 /* Rename lockres doesn't come from a slab so we call init
673 * once on it manually. */
674 ocfs2_lock_res_init_once(res);
675 ocfs2_build_lock_name(OCFS2_LOCK_TYPE_RENAME, 0, 0, res->l_name);
676 ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_RENAME,
677 &ocfs2_rename_lops, osb);
680 static void ocfs2_nfs_sync_lock_res_init(struct ocfs2_lock_res *res,
681 struct ocfs2_super *osb)
683 /* nfs_sync lockres doesn't come from a slab so we call init
684 * once on it manually. */
685 ocfs2_lock_res_init_once(res);
686 ocfs2_build_lock_name(OCFS2_LOCK_TYPE_NFS_SYNC, 0, 0, res->l_name);
687 ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_NFS_SYNC,
688 &ocfs2_nfs_sync_lops, osb);
691 static void ocfs2_nfs_sync_lock_init(struct ocfs2_super *osb)
693 ocfs2_nfs_sync_lock_res_init(&osb->osb_nfs_sync_lockres, osb);
694 init_rwsem(&osb->nfs_sync_rwlock);
697 void ocfs2_trim_fs_lock_res_init(struct ocfs2_super *osb)
699 struct ocfs2_lock_res *lockres = &osb->osb_trim_fs_lockres;
701 /* Only one trimfs thread are allowed to work at the same time. */
702 mutex_lock(&osb->obs_trim_fs_mutex);
704 ocfs2_lock_res_init_once(lockres);
705 ocfs2_build_lock_name(OCFS2_LOCK_TYPE_TRIM_FS, 0, 0, lockres->l_name);
706 ocfs2_lock_res_init_common(osb, lockres, OCFS2_LOCK_TYPE_TRIM_FS,
707 &ocfs2_trim_fs_lops, osb);
710 void ocfs2_trim_fs_lock_res_uninit(struct ocfs2_super *osb)
712 struct ocfs2_lock_res *lockres = &osb->osb_trim_fs_lockres;
714 ocfs2_simple_drop_lockres(osb, lockres);
715 ocfs2_lock_res_free(lockres);
717 mutex_unlock(&osb->obs_trim_fs_mutex);
720 static void ocfs2_orphan_scan_lock_res_init(struct ocfs2_lock_res *res,
721 struct ocfs2_super *osb)
723 ocfs2_lock_res_init_once(res);
724 ocfs2_build_lock_name(OCFS2_LOCK_TYPE_ORPHAN_SCAN, 0, 0, res->l_name);
725 ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_ORPHAN_SCAN,
726 &ocfs2_orphan_scan_lops, osb);
729 void ocfs2_file_lock_res_init(struct ocfs2_lock_res *lockres,
730 struct ocfs2_file_private *fp)
732 struct inode *inode = fp->fp_file->f_mapping->host;
733 struct ocfs2_inode_info *oi = OCFS2_I(inode);
735 ocfs2_lock_res_init_once(lockres);
736 ocfs2_build_lock_name(OCFS2_LOCK_TYPE_FLOCK, oi->ip_blkno,
737 inode->i_generation, lockres->l_name);
738 ocfs2_lock_res_init_common(OCFS2_SB(inode->i_sb), lockres,
739 OCFS2_LOCK_TYPE_FLOCK, &ocfs2_flock_lops,
741 lockres->l_flags |= OCFS2_LOCK_NOCACHE;
744 void ocfs2_qinfo_lock_res_init(struct ocfs2_lock_res *lockres,
745 struct ocfs2_mem_dqinfo *info)
747 ocfs2_lock_res_init_once(lockres);
748 ocfs2_build_lock_name(OCFS2_LOCK_TYPE_QINFO, info->dqi_gi.dqi_type,
750 ocfs2_lock_res_init_common(OCFS2_SB(info->dqi_gi.dqi_sb), lockres,
751 OCFS2_LOCK_TYPE_QINFO, &ocfs2_qinfo_lops,
755 void ocfs2_refcount_lock_res_init(struct ocfs2_lock_res *lockres,
756 struct ocfs2_super *osb, u64 ref_blkno,
757 unsigned int generation)
759 ocfs2_lock_res_init_once(lockres);
760 ocfs2_build_lock_name(OCFS2_LOCK_TYPE_REFCOUNT, ref_blkno,
761 generation, lockres->l_name);
762 ocfs2_lock_res_init_common(osb, lockres, OCFS2_LOCK_TYPE_REFCOUNT,
763 &ocfs2_refcount_block_lops, osb);
766 void ocfs2_lock_res_free(struct ocfs2_lock_res *res)
768 if (!(res->l_flags & OCFS2_LOCK_INITIALIZED))
771 ocfs2_remove_lockres_tracking(res);
773 mlog_bug_on_msg(!list_empty(&res->l_blocked_list),
774 "Lockres %s is on the blocked list\n",
776 mlog_bug_on_msg(!list_empty(&res->l_mask_waiters),
777 "Lockres %s has mask waiters pending\n",
779 mlog_bug_on_msg(spin_is_locked(&res->l_lock),
780 "Lockres %s is locked\n",
782 mlog_bug_on_msg(res->l_ro_holders,
783 "Lockres %s has %u ro holders\n",
784 res->l_name, res->l_ro_holders);
785 mlog_bug_on_msg(res->l_ex_holders,
786 "Lockres %s has %u ex holders\n",
787 res->l_name, res->l_ex_holders);
789 /* Need to clear out the lock status block for the dlm */
790 memset(&res->l_lksb, 0, sizeof(res->l_lksb));
796 * Keep a list of processes who have interest in a lockres.
797 * Note: this is now only uesed for check recursive cluster locking.
799 static inline void ocfs2_add_holder(struct ocfs2_lock_res *lockres,
800 struct ocfs2_lock_holder *oh)
802 INIT_LIST_HEAD(&oh->oh_list);
803 oh->oh_owner_pid = get_pid(task_pid(current));
805 spin_lock(&lockres->l_lock);
806 list_add_tail(&oh->oh_list, &lockres->l_holders);
807 spin_unlock(&lockres->l_lock);
810 static struct ocfs2_lock_holder *
811 ocfs2_pid_holder(struct ocfs2_lock_res *lockres,
814 struct ocfs2_lock_holder *oh;
816 spin_lock(&lockres->l_lock);
817 list_for_each_entry(oh, &lockres->l_holders, oh_list) {
818 if (oh->oh_owner_pid == pid) {
819 spin_unlock(&lockres->l_lock);
823 spin_unlock(&lockres->l_lock);
827 static inline void ocfs2_remove_holder(struct ocfs2_lock_res *lockres,
828 struct ocfs2_lock_holder *oh)
830 spin_lock(&lockres->l_lock);
831 list_del(&oh->oh_list);
832 spin_unlock(&lockres->l_lock);
834 put_pid(oh->oh_owner_pid);
838 static inline void ocfs2_inc_holders(struct ocfs2_lock_res *lockres,
845 lockres->l_ex_holders++;
848 lockres->l_ro_holders++;
855 static inline void ocfs2_dec_holders(struct ocfs2_lock_res *lockres,
862 BUG_ON(!lockres->l_ex_holders);
863 lockres->l_ex_holders--;
866 BUG_ON(!lockres->l_ro_holders);
867 lockres->l_ro_holders--;
874 /* WARNING: This function lives in a world where the only three lock
875 * levels are EX, PR, and NL. It *will* have to be adjusted when more
876 * lock types are added. */
877 static inline int ocfs2_highest_compat_lock_level(int level)
879 int new_level = DLM_LOCK_EX;
881 if (level == DLM_LOCK_EX)
882 new_level = DLM_LOCK_NL;
883 else if (level == DLM_LOCK_PR)
884 new_level = DLM_LOCK_PR;
888 static void lockres_set_flags(struct ocfs2_lock_res *lockres,
889 unsigned long newflags)
891 struct ocfs2_mask_waiter *mw, *tmp;
893 assert_spin_locked(&lockres->l_lock);
895 lockres->l_flags = newflags;
897 list_for_each_entry_safe(mw, tmp, &lockres->l_mask_waiters, mw_item) {
898 if ((lockres->l_flags & mw->mw_mask) != mw->mw_goal)
901 list_del_init(&mw->mw_item);
903 complete(&mw->mw_complete);
904 ocfs2_track_lock_wait(lockres);
907 static void lockres_or_flags(struct ocfs2_lock_res *lockres, unsigned long or)
909 lockres_set_flags(lockres, lockres->l_flags | or);
911 static void lockres_clear_flags(struct ocfs2_lock_res *lockres,
914 lockres_set_flags(lockres, lockres->l_flags & ~clear);
917 static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res *lockres)
919 BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
920 BUG_ON(!(lockres->l_flags & OCFS2_LOCK_ATTACHED));
921 BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BLOCKED));
922 BUG_ON(lockres->l_blocking <= DLM_LOCK_NL);
924 lockres->l_level = lockres->l_requested;
925 if (lockres->l_level <=
926 ocfs2_highest_compat_lock_level(lockres->l_blocking)) {
927 lockres->l_blocking = DLM_LOCK_NL;
928 lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED);
930 lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
933 static inline void ocfs2_generic_handle_convert_action(struct ocfs2_lock_res *lockres)
935 BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
936 BUG_ON(!(lockres->l_flags & OCFS2_LOCK_ATTACHED));
938 /* Convert from RO to EX doesn't really need anything as our
939 * information is already up to data. Convert from NL to
940 * *anything* however should mark ourselves as needing an
942 if (lockres->l_level == DLM_LOCK_NL &&
943 lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
944 lockres_or_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
946 lockres->l_level = lockres->l_requested;
949 * We set the OCFS2_LOCK_UPCONVERT_FINISHING flag before clearing
950 * the OCFS2_LOCK_BUSY flag to prevent the dc thread from
951 * downconverting the lock before the upconvert has fully completed.
952 * Do not prevent the dc thread from downconverting if NONBLOCK lock
953 * had already returned.
955 if (!(lockres->l_flags & OCFS2_LOCK_NONBLOCK_FINISHED))
956 lockres_or_flags(lockres, OCFS2_LOCK_UPCONVERT_FINISHING);
958 lockres_clear_flags(lockres, OCFS2_LOCK_NONBLOCK_FINISHED);
960 lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
963 static inline void ocfs2_generic_handle_attach_action(struct ocfs2_lock_res *lockres)
965 BUG_ON((!(lockres->l_flags & OCFS2_LOCK_BUSY)));
966 BUG_ON(lockres->l_flags & OCFS2_LOCK_ATTACHED);
968 if (lockres->l_requested > DLM_LOCK_NL &&
969 !(lockres->l_flags & OCFS2_LOCK_LOCAL) &&
970 lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
971 lockres_or_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
973 lockres->l_level = lockres->l_requested;
974 lockres_or_flags(lockres, OCFS2_LOCK_ATTACHED);
975 lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
978 static int ocfs2_generic_handle_bast(struct ocfs2_lock_res *lockres,
981 int needs_downconvert = 0;
983 assert_spin_locked(&lockres->l_lock);
985 if (level > lockres->l_blocking) {
986 /* only schedule a downconvert if we haven't already scheduled
987 * one that goes low enough to satisfy the level we're
988 * blocking. this also catches the case where we get
990 if (ocfs2_highest_compat_lock_level(level) <
991 ocfs2_highest_compat_lock_level(lockres->l_blocking))
992 needs_downconvert = 1;
994 lockres->l_blocking = level;
997 mlog(ML_BASTS, "lockres %s, block %d, level %d, l_block %d, dwn %d\n",
998 lockres->l_name, level, lockres->l_level, lockres->l_blocking,
1001 if (needs_downconvert)
1002 lockres_or_flags(lockres, OCFS2_LOCK_BLOCKED);
1003 mlog(0, "needs_downconvert = %d\n", needs_downconvert);
1004 return needs_downconvert;
1008 * OCFS2_LOCK_PENDING and l_pending_gen.
1010 * Why does OCFS2_LOCK_PENDING exist? To close a race between setting
1011 * OCFS2_LOCK_BUSY and calling ocfs2_dlm_lock(). See ocfs2_unblock_lock()
1012 * for more details on the race.
1014 * OCFS2_LOCK_PENDING closes the race quite nicely. However, it introduces
1015 * a race on itself. In o2dlm, we can get the ast before ocfs2_dlm_lock()
1016 * returns. The ast clears OCFS2_LOCK_BUSY, and must therefore clear
1017 * OCFS2_LOCK_PENDING at the same time. When ocfs2_dlm_lock() returns,
1018 * the caller is going to try to clear PENDING again. If nothing else is
1019 * happening, __lockres_clear_pending() sees PENDING is unset and does
1022 * But what if another path (eg downconvert thread) has just started a
1023 * new locking action? The other path has re-set PENDING. Our path
1024 * cannot clear PENDING, because that will re-open the original race
1030 * ocfs2_cluster_lock()
1035 * ocfs2_locking_ast() ocfs2_downconvert_thread()
1036 * clear PENDING ocfs2_unblock_lock()
1039 * ocfs2_prepare_downconvert()
1049 * So as you can see, we now have a window where l_lock is not held,
1050 * PENDING is not set, and ocfs2_dlm_lock() has not been called.
1052 * The core problem is that ocfs2_cluster_lock() has cleared the PENDING
1053 * set by ocfs2_prepare_downconvert(). That wasn't nice.
1055 * To solve this we introduce l_pending_gen. A call to
1056 * lockres_clear_pending() will only do so when it is passed a generation
1057 * number that matches the lockres. lockres_set_pending() will return the
1058 * current generation number. When ocfs2_cluster_lock() goes to clear
1059 * PENDING, it passes the generation it got from set_pending(). In our
1060 * example above, the generation numbers will *not* match. Thus,
1061 * ocfs2_cluster_lock() will not clear the PENDING set by
1062 * ocfs2_prepare_downconvert().
1065 /* Unlocked version for ocfs2_locking_ast() */
1066 static void __lockres_clear_pending(struct ocfs2_lock_res *lockres,
1067 unsigned int generation,
1068 struct ocfs2_super *osb)
1070 assert_spin_locked(&lockres->l_lock);
1073 * The ast and locking functions can race us here. The winner
1074 * will clear pending, the loser will not.
1076 if (!(lockres->l_flags & OCFS2_LOCK_PENDING) ||
1077 (lockres->l_pending_gen != generation))
1080 lockres_clear_flags(lockres, OCFS2_LOCK_PENDING);
1081 lockres->l_pending_gen++;
1084 * The downconvert thread may have skipped us because we
1085 * were PENDING. Wake it up.
1087 if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
1088 ocfs2_wake_downconvert_thread(osb);
1091 /* Locked version for callers of ocfs2_dlm_lock() */
1092 static void lockres_clear_pending(struct ocfs2_lock_res *lockres,
1093 unsigned int generation,
1094 struct ocfs2_super *osb)
1096 unsigned long flags;
1098 spin_lock_irqsave(&lockres->l_lock, flags);
1099 __lockres_clear_pending(lockres, generation, osb);
1100 spin_unlock_irqrestore(&lockres->l_lock, flags);
1103 static unsigned int lockres_set_pending(struct ocfs2_lock_res *lockres)
1105 assert_spin_locked(&lockres->l_lock);
1106 BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
1108 lockres_or_flags(lockres, OCFS2_LOCK_PENDING);
1110 return lockres->l_pending_gen;
1113 static void ocfs2_blocking_ast(struct ocfs2_dlm_lksb *lksb, int level)
1115 struct ocfs2_lock_res *lockres = ocfs2_lksb_to_lock_res(lksb);
1116 struct ocfs2_super *osb = ocfs2_get_lockres_osb(lockres);
1117 int needs_downconvert;
1118 unsigned long flags;
1120 BUG_ON(level <= DLM_LOCK_NL);
1122 mlog(ML_BASTS, "BAST fired for lockres %s, blocking %d, level %d, "
1123 "type %s\n", lockres->l_name, level, lockres->l_level,
1124 ocfs2_lock_type_string(lockres->l_type));
1127 * We can skip the bast for locks which don't enable caching -
1128 * they'll be dropped at the earliest possible time anyway.
1130 if (lockres->l_flags & OCFS2_LOCK_NOCACHE)
1133 spin_lock_irqsave(&lockres->l_lock, flags);
1134 needs_downconvert = ocfs2_generic_handle_bast(lockres, level);
1135 if (needs_downconvert)
1136 ocfs2_schedule_blocked_lock(osb, lockres);
1137 spin_unlock_irqrestore(&lockres->l_lock, flags);
1139 wake_up(&lockres->l_event);
1141 ocfs2_wake_downconvert_thread(osb);
1144 static void ocfs2_locking_ast(struct ocfs2_dlm_lksb *lksb)
1146 struct ocfs2_lock_res *lockres = ocfs2_lksb_to_lock_res(lksb);
1147 struct ocfs2_super *osb = ocfs2_get_lockres_osb(lockres);
1148 unsigned long flags;
1151 spin_lock_irqsave(&lockres->l_lock, flags);
1153 status = ocfs2_dlm_lock_status(&lockres->l_lksb);
1155 if (status == -EAGAIN) {
1156 lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
1161 mlog(ML_ERROR, "lockres %s: lksb status value of %d!\n",
1162 lockres->l_name, status);
1163 spin_unlock_irqrestore(&lockres->l_lock, flags);
1167 mlog(ML_BASTS, "AST fired for lockres %s, action %d, unlock %d, "
1168 "level %d => %d\n", lockres->l_name, lockres->l_action,
1169 lockres->l_unlock_action, lockres->l_level, lockres->l_requested);
1171 switch(lockres->l_action) {
1172 case OCFS2_AST_ATTACH:
1173 ocfs2_generic_handle_attach_action(lockres);
1174 lockres_clear_flags(lockres, OCFS2_LOCK_LOCAL);
1176 case OCFS2_AST_CONVERT:
1177 ocfs2_generic_handle_convert_action(lockres);
1179 case OCFS2_AST_DOWNCONVERT:
1180 ocfs2_generic_handle_downconvert_action(lockres);
1183 mlog(ML_ERROR, "lockres %s: AST fired with invalid action: %u, "
1184 "flags 0x%lx, unlock: %u\n",
1185 lockres->l_name, lockres->l_action, lockres->l_flags,
1186 lockres->l_unlock_action);
1190 /* set it to something invalid so if we get called again we
1192 lockres->l_action = OCFS2_AST_INVALID;
1194 /* Did we try to cancel this lock? Clear that state */
1195 if (lockres->l_unlock_action == OCFS2_UNLOCK_CANCEL_CONVERT)
1196 lockres->l_unlock_action = OCFS2_UNLOCK_INVALID;
1199 * We may have beaten the locking functions here. We certainly
1200 * know that dlm_lock() has been called :-)
1201 * Because we can't have two lock calls in flight at once, we
1202 * can use lockres->l_pending_gen.
1204 __lockres_clear_pending(lockres, lockres->l_pending_gen, osb);
1206 wake_up(&lockres->l_event);
1207 spin_unlock_irqrestore(&lockres->l_lock, flags);
1210 static void ocfs2_unlock_ast(struct ocfs2_dlm_lksb *lksb, int error)
1212 struct ocfs2_lock_res *lockres = ocfs2_lksb_to_lock_res(lksb);
1213 unsigned long flags;
1215 mlog(ML_BASTS, "UNLOCK AST fired for lockres %s, action = %d\n",
1216 lockres->l_name, lockres->l_unlock_action);
1218 spin_lock_irqsave(&lockres->l_lock, flags);
1220 mlog(ML_ERROR, "Dlm passes error %d for lock %s, "
1221 "unlock_action %d\n", error, lockres->l_name,
1222 lockres->l_unlock_action);
1223 spin_unlock_irqrestore(&lockres->l_lock, flags);
1227 switch(lockres->l_unlock_action) {
1228 case OCFS2_UNLOCK_CANCEL_CONVERT:
1229 mlog(0, "Cancel convert success for %s\n", lockres->l_name);
1230 lockres->l_action = OCFS2_AST_INVALID;
1231 /* Downconvert thread may have requeued this lock, we
1232 * need to wake it. */
1233 if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
1234 ocfs2_wake_downconvert_thread(ocfs2_get_lockres_osb(lockres));
1236 case OCFS2_UNLOCK_DROP_LOCK:
1237 lockres->l_level = DLM_LOCK_IV;
1243 lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
1244 lockres->l_unlock_action = OCFS2_UNLOCK_INVALID;
1245 wake_up(&lockres->l_event);
1246 spin_unlock_irqrestore(&lockres->l_lock, flags);
1250 * This is the filesystem locking protocol. It provides the lock handling
1251 * hooks for the underlying DLM. It has a maximum version number.
1252 * The version number allows interoperability with systems running at
1253 * the same major number and an equal or smaller minor number.
1255 * Whenever the filesystem does new things with locks (adds or removes a
1256 * lock, orders them differently, does different things underneath a lock),
1257 * the version must be changed. The protocol is negotiated when joining
1258 * the dlm domain. A node may join the domain if its major version is
1259 * identical to all other nodes and its minor version is greater than
1260 * or equal to all other nodes. When its minor version is greater than
1261 * the other nodes, it will run at the minor version specified by the
1264 * If a locking change is made that will not be compatible with older
1265 * versions, the major number must be increased and the minor version set
1266 * to zero. If a change merely adds a behavior that can be disabled when
1267 * speaking to older versions, the minor version must be increased. If a
1268 * change adds a fully backwards compatible change (eg, LVB changes that
1269 * are just ignored by older versions), the version does not need to be
1272 static struct ocfs2_locking_protocol lproto = {
1274 .pv_major = OCFS2_LOCKING_PROTOCOL_MAJOR,
1275 .pv_minor = OCFS2_LOCKING_PROTOCOL_MINOR,
1277 .lp_lock_ast = ocfs2_locking_ast,
1278 .lp_blocking_ast = ocfs2_blocking_ast,
1279 .lp_unlock_ast = ocfs2_unlock_ast,
1282 void ocfs2_set_locking_protocol(void)
1284 ocfs2_stack_glue_set_max_proto_version(&lproto.lp_max_version);
1287 static inline void ocfs2_recover_from_dlm_error(struct ocfs2_lock_res *lockres,
1290 unsigned long flags;
1292 spin_lock_irqsave(&lockres->l_lock, flags);
1293 lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
1294 lockres_clear_flags(lockres, OCFS2_LOCK_UPCONVERT_FINISHING);
1296 lockres->l_action = OCFS2_AST_INVALID;
1298 lockres->l_unlock_action = OCFS2_UNLOCK_INVALID;
1299 spin_unlock_irqrestore(&lockres->l_lock, flags);
1301 wake_up(&lockres->l_event);
1304 /* Note: If we detect another process working on the lock (i.e.,
1305 * OCFS2_LOCK_BUSY), we'll bail out returning 0. It's up to the caller
1306 * to do the right thing in that case.
1308 static int ocfs2_lock_create(struct ocfs2_super *osb,
1309 struct ocfs2_lock_res *lockres,
1314 unsigned long flags;
1317 mlog(0, "lock %s, level = %d, flags = %u\n", lockres->l_name, level,
1320 spin_lock_irqsave(&lockres->l_lock, flags);
1321 if ((lockres->l_flags & OCFS2_LOCK_ATTACHED) ||
1322 (lockres->l_flags & OCFS2_LOCK_BUSY)) {
1323 spin_unlock_irqrestore(&lockres->l_lock, flags);
1327 lockres->l_action = OCFS2_AST_ATTACH;
1328 lockres->l_requested = level;
1329 lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
1330 gen = lockres_set_pending(lockres);
1331 spin_unlock_irqrestore(&lockres->l_lock, flags);
1333 ret = ocfs2_dlm_lock(osb->cconn,
1338 OCFS2_LOCK_ID_MAX_LEN - 1);
1339 lockres_clear_pending(lockres, gen, osb);
1341 ocfs2_log_dlm_error("ocfs2_dlm_lock", ret, lockres);
1342 ocfs2_recover_from_dlm_error(lockres, 1);
1345 mlog(0, "lock %s, return from ocfs2_dlm_lock\n", lockres->l_name);
1351 static inline int ocfs2_check_wait_flag(struct ocfs2_lock_res *lockres,
1354 unsigned long flags;
1357 spin_lock_irqsave(&lockres->l_lock, flags);
1358 ret = lockres->l_flags & flag;
1359 spin_unlock_irqrestore(&lockres->l_lock, flags);
1364 static inline void ocfs2_wait_on_busy_lock(struct ocfs2_lock_res *lockres)
1367 wait_event(lockres->l_event,
1368 !ocfs2_check_wait_flag(lockres, OCFS2_LOCK_BUSY));
1371 static inline void ocfs2_wait_on_refreshing_lock(struct ocfs2_lock_res *lockres)
1374 wait_event(lockres->l_event,
1375 !ocfs2_check_wait_flag(lockres, OCFS2_LOCK_REFRESHING));
1378 /* predict what lock level we'll be dropping down to on behalf
1379 * of another node, and return true if the currently wanted
1380 * level will be compatible with it. */
1381 static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lockres,
1384 BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BLOCKED));
1386 return wanted <= ocfs2_highest_compat_lock_level(lockres->l_blocking);
1389 static void ocfs2_init_mask_waiter(struct ocfs2_mask_waiter *mw)
1391 INIT_LIST_HEAD(&mw->mw_item);
1392 init_completion(&mw->mw_complete);
1393 ocfs2_init_start_time(mw);
1396 static int ocfs2_wait_for_mask(struct ocfs2_mask_waiter *mw)
1398 wait_for_completion(&mw->mw_complete);
1399 /* Re-arm the completion in case we want to wait on it again */
1400 reinit_completion(&mw->mw_complete);
1401 return mw->mw_status;
1404 static void lockres_add_mask_waiter(struct ocfs2_lock_res *lockres,
1405 struct ocfs2_mask_waiter *mw,
1409 BUG_ON(!list_empty(&mw->mw_item));
1411 assert_spin_locked(&lockres->l_lock);
1413 list_add_tail(&mw->mw_item, &lockres->l_mask_waiters);
1416 ocfs2_track_lock_wait(lockres);
1419 /* returns 0 if the mw that was removed was already satisfied, -EBUSY
1420 * if the mask still hadn't reached its goal */
1421 static int __lockres_remove_mask_waiter(struct ocfs2_lock_res *lockres,
1422 struct ocfs2_mask_waiter *mw)
1426 assert_spin_locked(&lockres->l_lock);
1427 if (!list_empty(&mw->mw_item)) {
1428 if ((lockres->l_flags & mw->mw_mask) != mw->mw_goal)
1431 list_del_init(&mw->mw_item);
1432 init_completion(&mw->mw_complete);
1433 ocfs2_track_lock_wait(lockres);
1439 static int lockres_remove_mask_waiter(struct ocfs2_lock_res *lockres,
1440 struct ocfs2_mask_waiter *mw)
1442 unsigned long flags;
1445 spin_lock_irqsave(&lockres->l_lock, flags);
1446 ret = __lockres_remove_mask_waiter(lockres, mw);
1447 spin_unlock_irqrestore(&lockres->l_lock, flags);
1453 static int ocfs2_wait_for_mask_interruptible(struct ocfs2_mask_waiter *mw,
1454 struct ocfs2_lock_res *lockres)
1458 ret = wait_for_completion_interruptible(&mw->mw_complete);
1460 lockres_remove_mask_waiter(lockres, mw);
1462 ret = mw->mw_status;
1463 /* Re-arm the completion in case we want to wait on it again */
1464 reinit_completion(&mw->mw_complete);
1468 static int __ocfs2_cluster_lock(struct ocfs2_super *osb,
1469 struct ocfs2_lock_res *lockres,
1474 unsigned long caller_ip)
1476 struct ocfs2_mask_waiter mw;
1477 int wait, catch_signals = !(osb->s_mount_opt & OCFS2_MOUNT_NOINTR);
1478 int ret = 0; /* gcc doesn't realize wait = 1 guarantees ret is set */
1479 unsigned long flags;
1481 int noqueue_attempted = 0;
1485 if (!(lockres->l_flags & OCFS2_LOCK_INITIALIZED)) {
1486 mlog_errno(-EINVAL);
1490 ocfs2_init_mask_waiter(&mw);
1492 if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
1493 lkm_flags |= DLM_LKF_VALBLK;
1498 spin_lock_irqsave(&lockres->l_lock, flags);
1500 if (catch_signals && signal_pending(current)) {
1505 mlog_bug_on_msg(lockres->l_flags & OCFS2_LOCK_FREEING,
1506 "Cluster lock called on freeing lockres %s! flags "
1507 "0x%lx\n", lockres->l_name, lockres->l_flags);
1509 /* We only compare against the currently granted level
1510 * here. If the lock is blocked waiting on a downconvert,
1511 * we'll get caught below. */
1512 if (lockres->l_flags & OCFS2_LOCK_BUSY &&
1513 level > lockres->l_level) {
1514 /* is someone sitting in dlm_lock? If so, wait on
1516 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1521 if (lockres->l_flags & OCFS2_LOCK_UPCONVERT_FINISHING) {
1523 * We've upconverted. If the lock now has a level we can
1524 * work with, we take it. If, however, the lock is not at the
1525 * required level, we go thru the full cycle. One way this could
1526 * happen is if a process requesting an upconvert to PR is
1527 * closely followed by another requesting upconvert to an EX.
1528 * If the process requesting EX lands here, we want it to
1529 * continue attempting to upconvert and let the process
1530 * requesting PR take the lock.
1531 * If multiple processes request upconvert to PR, the first one
1532 * here will take the lock. The others will have to go thru the
1533 * OCFS2_LOCK_BLOCKED check to ensure that there is no pending
1534 * downconvert request.
1536 if (level <= lockres->l_level)
1537 goto update_holders;
1540 if (lockres->l_flags & OCFS2_LOCK_BLOCKED &&
1541 !ocfs2_may_continue_on_blocked_lock(lockres, level)) {
1542 /* is the lock is currently blocked on behalf of
1544 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BLOCKED, 0);
1549 if (level > lockres->l_level) {
1550 if (noqueue_attempted > 0) {
1554 if (lkm_flags & DLM_LKF_NOQUEUE)
1555 noqueue_attempted = 1;
1557 if (lockres->l_action != OCFS2_AST_INVALID)
1558 mlog(ML_ERROR, "lockres %s has action %u pending\n",
1559 lockres->l_name, lockres->l_action);
1561 if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
1562 lockres->l_action = OCFS2_AST_ATTACH;
1563 lkm_flags &= ~DLM_LKF_CONVERT;
1565 lockres->l_action = OCFS2_AST_CONVERT;
1566 lkm_flags |= DLM_LKF_CONVERT;
1569 lockres->l_requested = level;
1570 lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
1571 gen = lockres_set_pending(lockres);
1572 spin_unlock_irqrestore(&lockres->l_lock, flags);
1574 BUG_ON(level == DLM_LOCK_IV);
1575 BUG_ON(level == DLM_LOCK_NL);
1577 mlog(ML_BASTS, "lockres %s, convert from %d to %d\n",
1578 lockres->l_name, lockres->l_level, level);
1580 /* call dlm_lock to upgrade lock now */
1581 ret = ocfs2_dlm_lock(osb->cconn,
1586 OCFS2_LOCK_ID_MAX_LEN - 1);
1587 lockres_clear_pending(lockres, gen, osb);
1589 if (!(lkm_flags & DLM_LKF_NOQUEUE) ||
1591 ocfs2_log_dlm_error("ocfs2_dlm_lock",
1594 ocfs2_recover_from_dlm_error(lockres, 1);
1599 mlog(0, "lock %s, successful return from ocfs2_dlm_lock\n",
1602 /* At this point we've gone inside the dlm and need to
1603 * complete our work regardless. */
1606 /* wait for busy to clear and carry on */
1611 /* Ok, if we get here then we're good to go. */
1612 ocfs2_inc_holders(lockres, level);
1616 lockres_clear_flags(lockres, OCFS2_LOCK_UPCONVERT_FINISHING);
1618 /* ocfs2_unblock_lock reques on seeing OCFS2_LOCK_UPCONVERT_FINISHING */
1619 kick_dc = (lockres->l_flags & OCFS2_LOCK_BLOCKED);
1621 spin_unlock_irqrestore(&lockres->l_lock, flags);
1623 ocfs2_wake_downconvert_thread(osb);
1626 * This is helping work around a lock inversion between the page lock
1627 * and dlm locks. One path holds the page lock while calling aops
1628 * which block acquiring dlm locks. The voting thread holds dlm
1629 * locks while acquiring page locks while down converting data locks.
1630 * This block is helping an aop path notice the inversion and back
1631 * off to unlock its page lock before trying the dlm lock again.
1633 if (wait && arg_flags & OCFS2_LOCK_NONBLOCK &&
1634 mw.mw_mask & (OCFS2_LOCK_BUSY|OCFS2_LOCK_BLOCKED)) {
1636 spin_lock_irqsave(&lockres->l_lock, flags);
1637 if (__lockres_remove_mask_waiter(lockres, &mw)) {
1639 lockres_or_flags(lockres,
1640 OCFS2_LOCK_NONBLOCK_FINISHED);
1641 spin_unlock_irqrestore(&lockres->l_lock, flags);
1644 spin_unlock_irqrestore(&lockres->l_lock, flags);
1649 ret = ocfs2_wait_for_mask(&mw);
1654 ocfs2_update_lock_stats(lockres, level, &mw, ret);
1656 #ifdef CONFIG_DEBUG_LOCK_ALLOC
1657 if (!ret && lockres->l_lockdep_map.key != NULL) {
1658 if (level == DLM_LOCK_PR)
1659 rwsem_acquire_read(&lockres->l_lockdep_map, l_subclass,
1660 !!(arg_flags & OCFS2_META_LOCK_NOQUEUE),
1663 rwsem_acquire(&lockres->l_lockdep_map, l_subclass,
1664 !!(arg_flags & OCFS2_META_LOCK_NOQUEUE),
1671 static inline int ocfs2_cluster_lock(struct ocfs2_super *osb,
1672 struct ocfs2_lock_res *lockres,
1677 return __ocfs2_cluster_lock(osb, lockres, level, lkm_flags, arg_flags,
1682 static void __ocfs2_cluster_unlock(struct ocfs2_super *osb,
1683 struct ocfs2_lock_res *lockres,
1685 unsigned long caller_ip)
1687 unsigned long flags;
1689 spin_lock_irqsave(&lockres->l_lock, flags);
1690 ocfs2_dec_holders(lockres, level);
1691 ocfs2_downconvert_on_unlock(osb, lockres);
1692 spin_unlock_irqrestore(&lockres->l_lock, flags);
1693 #ifdef CONFIG_DEBUG_LOCK_ALLOC
1694 if (lockres->l_lockdep_map.key != NULL)
1695 rwsem_release(&lockres->l_lockdep_map, caller_ip);
1699 static int ocfs2_create_new_lock(struct ocfs2_super *osb,
1700 struct ocfs2_lock_res *lockres,
1704 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
1705 unsigned long flags;
1706 u32 lkm_flags = local ? DLM_LKF_LOCAL : 0;
1708 spin_lock_irqsave(&lockres->l_lock, flags);
1709 BUG_ON(lockres->l_flags & OCFS2_LOCK_ATTACHED);
1710 lockres_or_flags(lockres, OCFS2_LOCK_LOCAL);
1711 spin_unlock_irqrestore(&lockres->l_lock, flags);
1713 return ocfs2_lock_create(osb, lockres, level, lkm_flags);
1716 /* Grants us an EX lock on the data and metadata resources, skipping
1717 * the normal cluster directory lookup. Use this ONLY on newly created
1718 * inodes which other nodes can't possibly see, and which haven't been
1719 * hashed in the inode hash yet. This can give us a good performance
1720 * increase as it'll skip the network broadcast normally associated
1721 * with creating a new lock resource. */
1722 int ocfs2_create_new_inode_locks(struct inode *inode)
1725 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1727 BUG_ON(!ocfs2_inode_is_new(inode));
1729 mlog(0, "Inode %llu\n", (unsigned long long)OCFS2_I(inode)->ip_blkno);
1731 /* NOTE: That we don't increment any of the holder counts, nor
1732 * do we add anything to a journal handle. Since this is
1733 * supposed to be a new inode which the cluster doesn't know
1734 * about yet, there is no need to. As far as the LVB handling
1735 * is concerned, this is basically like acquiring an EX lock
1736 * on a resource which has an invalid one -- we'll set it
1737 * valid when we release the EX. */
1739 ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_rw_lockres, 1, 1);
1746 * We don't want to use DLM_LKF_LOCAL on a meta data lock as they
1747 * don't use a generation in their lock names.
1749 ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_inode_lockres, 1, 0);
1755 ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_open_lockres, 0, 0);
1763 int ocfs2_rw_lock(struct inode *inode, int write)
1766 struct ocfs2_lock_res *lockres;
1767 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1769 mlog(0, "inode %llu take %s RW lock\n",
1770 (unsigned long long)OCFS2_I(inode)->ip_blkno,
1771 write ? "EXMODE" : "PRMODE");
1773 if (ocfs2_mount_local(osb))
1776 lockres = &OCFS2_I(inode)->ip_rw_lockres;
1778 level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1780 status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
1787 int ocfs2_try_rw_lock(struct inode *inode, int write)
1790 struct ocfs2_lock_res *lockres;
1791 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1793 mlog(0, "inode %llu try to take %s RW lock\n",
1794 (unsigned long long)OCFS2_I(inode)->ip_blkno,
1795 write ? "EXMODE" : "PRMODE");
1797 if (ocfs2_mount_local(osb))
1800 lockres = &OCFS2_I(inode)->ip_rw_lockres;
1802 level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1804 status = ocfs2_cluster_lock(osb, lockres, level, DLM_LKF_NOQUEUE, 0);
1808 void ocfs2_rw_unlock(struct inode *inode, int write)
1810 int level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1811 struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_rw_lockres;
1812 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1814 mlog(0, "inode %llu drop %s RW lock\n",
1815 (unsigned long long)OCFS2_I(inode)->ip_blkno,
1816 write ? "EXMODE" : "PRMODE");
1818 if (!ocfs2_mount_local(osb))
1819 ocfs2_cluster_unlock(osb, lockres, level);
1823 * ocfs2_open_lock always get PR mode lock.
1825 int ocfs2_open_lock(struct inode *inode)
1828 struct ocfs2_lock_res *lockres;
1829 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1831 mlog(0, "inode %llu take PRMODE open lock\n",
1832 (unsigned long long)OCFS2_I(inode)->ip_blkno);
1834 if (ocfs2_is_hard_readonly(osb) || ocfs2_mount_local(osb))
1837 lockres = &OCFS2_I(inode)->ip_open_lockres;
1839 status = ocfs2_cluster_lock(osb, lockres, DLM_LOCK_PR, 0, 0);
1847 int ocfs2_try_open_lock(struct inode *inode, int write)
1849 int status = 0, level;
1850 struct ocfs2_lock_res *lockres;
1851 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1853 mlog(0, "inode %llu try to take %s open lock\n",
1854 (unsigned long long)OCFS2_I(inode)->ip_blkno,
1855 write ? "EXMODE" : "PRMODE");
1857 if (ocfs2_is_hard_readonly(osb)) {
1863 if (ocfs2_mount_local(osb))
1866 lockres = &OCFS2_I(inode)->ip_open_lockres;
1868 level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1871 * The file system may already holding a PRMODE/EXMODE open lock.
1872 * Since we pass DLM_LKF_NOQUEUE, the request won't block waiting on
1873 * other nodes and the -EAGAIN will indicate to the caller that
1874 * this inode is still in use.
1876 status = ocfs2_cluster_lock(osb, lockres, level, DLM_LKF_NOQUEUE, 0);
1883 * ocfs2_open_unlock unlock PR and EX mode open locks.
1885 void ocfs2_open_unlock(struct inode *inode)
1887 struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_open_lockres;
1888 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1890 mlog(0, "inode %llu drop open lock\n",
1891 (unsigned long long)OCFS2_I(inode)->ip_blkno);
1893 if (ocfs2_mount_local(osb))
1896 if(lockres->l_ro_holders)
1897 ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_PR);
1898 if(lockres->l_ex_holders)
1899 ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
1905 static int ocfs2_flock_handle_signal(struct ocfs2_lock_res *lockres,
1909 struct ocfs2_super *osb = ocfs2_get_lockres_osb(lockres);
1910 unsigned long flags;
1911 struct ocfs2_mask_waiter mw;
1913 ocfs2_init_mask_waiter(&mw);
1916 spin_lock_irqsave(&lockres->l_lock, flags);
1917 if (lockres->l_flags & OCFS2_LOCK_BUSY) {
1918 ret = ocfs2_prepare_cancel_convert(osb, lockres);
1920 spin_unlock_irqrestore(&lockres->l_lock, flags);
1921 ret = ocfs2_cancel_convert(osb, lockres);
1928 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1929 spin_unlock_irqrestore(&lockres->l_lock, flags);
1931 ocfs2_wait_for_mask(&mw);
1937 * We may still have gotten the lock, in which case there's no
1938 * point to restarting the syscall.
1940 if (lockres->l_level == level)
1943 mlog(0, "Cancel returning %d. flags: 0x%lx, level: %d, act: %d\n", ret,
1944 lockres->l_flags, lockres->l_level, lockres->l_action);
1946 spin_unlock_irqrestore(&lockres->l_lock, flags);
1953 * ocfs2_file_lock() and ocfs2_file_unlock() map to a single pair of
1954 * flock() calls. The locking approach this requires is sufficiently
1955 * different from all other cluster lock types that we implement a
1956 * separate path to the "low-level" dlm calls. In particular:
1958 * - No optimization of lock levels is done - we take at exactly
1959 * what's been requested.
1961 * - No lock caching is employed. We immediately downconvert to
1962 * no-lock at unlock time. This also means flock locks never go on
1963 * the blocking list).
1965 * - Since userspace can trivially deadlock itself with flock, we make
1966 * sure to allow cancellation of a misbehaving applications flock()
1969 * - Access to any flock lockres doesn't require concurrency, so we
1970 * can simplify the code by requiring the caller to guarantee
1971 * serialization of dlmglue flock calls.
1973 int ocfs2_file_lock(struct file *file, int ex, int trylock)
1975 int ret, level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
1976 unsigned int lkm_flags = trylock ? DLM_LKF_NOQUEUE : 0;
1977 unsigned long flags;
1978 struct ocfs2_file_private *fp = file->private_data;
1979 struct ocfs2_lock_res *lockres = &fp->fp_flock;
1980 struct ocfs2_super *osb = OCFS2_SB(file->f_mapping->host->i_sb);
1981 struct ocfs2_mask_waiter mw;
1983 ocfs2_init_mask_waiter(&mw);
1985 if ((lockres->l_flags & OCFS2_LOCK_BUSY) ||
1986 (lockres->l_level > DLM_LOCK_NL)) {
1988 "File lock \"%s\" has busy or locked state: flags: 0x%lx, "
1989 "level: %u\n", lockres->l_name, lockres->l_flags,
1994 spin_lock_irqsave(&lockres->l_lock, flags);
1995 if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
1996 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1997 spin_unlock_irqrestore(&lockres->l_lock, flags);
2000 * Get the lock at NLMODE to start - that way we
2001 * can cancel the upconvert request if need be.
2003 ret = ocfs2_lock_create(osb, lockres, DLM_LOCK_NL, 0);
2009 ret = ocfs2_wait_for_mask(&mw);
2014 spin_lock_irqsave(&lockres->l_lock, flags);
2017 lockres->l_action = OCFS2_AST_CONVERT;
2018 lkm_flags |= DLM_LKF_CONVERT;
2019 lockres->l_requested = level;
2020 lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
2022 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
2023 spin_unlock_irqrestore(&lockres->l_lock, flags);
2025 ret = ocfs2_dlm_lock(osb->cconn, level, &lockres->l_lksb, lkm_flags,
2026 lockres->l_name, OCFS2_LOCK_ID_MAX_LEN - 1);
2028 if (!trylock || (ret != -EAGAIN)) {
2029 ocfs2_log_dlm_error("ocfs2_dlm_lock", ret, lockres);
2033 ocfs2_recover_from_dlm_error(lockres, 1);
2034 lockres_remove_mask_waiter(lockres, &mw);
2038 ret = ocfs2_wait_for_mask_interruptible(&mw, lockres);
2039 if (ret == -ERESTARTSYS) {
2041 * Userspace can cause deadlock itself with
2042 * flock(). Current behavior locally is to allow the
2043 * deadlock, but abort the system call if a signal is
2044 * received. We follow this example, otherwise a
2045 * poorly written program could sit in kernel until
2048 * Handling this is a bit more complicated for Ocfs2
2049 * though. We can't exit this function with an
2050 * outstanding lock request, so a cancel convert is
2051 * required. We intentionally overwrite 'ret' - if the
2052 * cancel fails and the lock was granted, it's easier
2053 * to just bubble success back up to the user.
2055 ret = ocfs2_flock_handle_signal(lockres, level);
2056 } else if (!ret && (level > lockres->l_level)) {
2057 /* Trylock failed asynchronously */
2064 mlog(0, "Lock: \"%s\" ex: %d, trylock: %d, returns: %d\n",
2065 lockres->l_name, ex, trylock, ret);
2069 void ocfs2_file_unlock(struct file *file)
2073 unsigned long flags;
2074 struct ocfs2_file_private *fp = file->private_data;
2075 struct ocfs2_lock_res *lockres = &fp->fp_flock;
2076 struct ocfs2_super *osb = OCFS2_SB(file->f_mapping->host->i_sb);
2077 struct ocfs2_mask_waiter mw;
2079 ocfs2_init_mask_waiter(&mw);
2081 if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED))
2084 if (lockres->l_level == DLM_LOCK_NL)
2087 mlog(0, "Unlock: \"%s\" flags: 0x%lx, level: %d, act: %d\n",
2088 lockres->l_name, lockres->l_flags, lockres->l_level,
2091 spin_lock_irqsave(&lockres->l_lock, flags);
2093 * Fake a blocking ast for the downconvert code.
2095 lockres_or_flags(lockres, OCFS2_LOCK_BLOCKED);
2096 lockres->l_blocking = DLM_LOCK_EX;
2098 gen = ocfs2_prepare_downconvert(lockres, DLM_LOCK_NL);
2099 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
2100 spin_unlock_irqrestore(&lockres->l_lock, flags);
2102 ret = ocfs2_downconvert_lock(osb, lockres, DLM_LOCK_NL, 0, gen);
2108 ret = ocfs2_wait_for_mask(&mw);
2113 static void ocfs2_downconvert_on_unlock(struct ocfs2_super *osb,
2114 struct ocfs2_lock_res *lockres)
2118 /* If we know that another node is waiting on our lock, kick
2119 * the downconvert thread * pre-emptively when we reach a release
2121 if (lockres->l_flags & OCFS2_LOCK_BLOCKED) {
2122 switch(lockres->l_blocking) {
2124 if (!lockres->l_ex_holders && !lockres->l_ro_holders)
2128 if (!lockres->l_ex_holders)
2137 ocfs2_wake_downconvert_thread(osb);
2140 #define OCFS2_SEC_BITS 34
2141 #define OCFS2_SEC_SHIFT (64 - OCFS2_SEC_BITS)
2142 #define OCFS2_NSEC_MASK ((1ULL << OCFS2_SEC_SHIFT) - 1)
2144 /* LVB only has room for 64 bits of time here so we pack it for
2146 static u64 ocfs2_pack_timespec(struct timespec64 *spec)
2149 u64 sec = clamp_t(time64_t, spec->tv_sec, 0, 0x3ffffffffull);
2150 u32 nsec = spec->tv_nsec;
2152 res = (sec << OCFS2_SEC_SHIFT) | (nsec & OCFS2_NSEC_MASK);
2157 /* Call this with the lockres locked. I am reasonably sure we don't
2158 * need ip_lock in this function as anyone who would be changing those
2159 * values is supposed to be blocked in ocfs2_inode_lock right now. */
2160 static void __ocfs2_stuff_meta_lvb(struct inode *inode)
2162 struct ocfs2_inode_info *oi = OCFS2_I(inode);
2163 struct ocfs2_lock_res *lockres = &oi->ip_inode_lockres;
2164 struct ocfs2_meta_lvb *lvb;
2166 lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2169 * Invalidate the LVB of a deleted inode - this way other
2170 * nodes are forced to go to disk and discover the new inode
2173 if (oi->ip_flags & OCFS2_INODE_DELETED) {
2174 lvb->lvb_version = 0;
2178 lvb->lvb_version = OCFS2_LVB_VERSION;
2179 lvb->lvb_isize = cpu_to_be64(i_size_read(inode));
2180 lvb->lvb_iclusters = cpu_to_be32(oi->ip_clusters);
2181 lvb->lvb_iuid = cpu_to_be32(i_uid_read(inode));
2182 lvb->lvb_igid = cpu_to_be32(i_gid_read(inode));
2183 lvb->lvb_imode = cpu_to_be16(inode->i_mode);
2184 lvb->lvb_inlink = cpu_to_be16(inode->i_nlink);
2185 lvb->lvb_iatime_packed =
2186 cpu_to_be64(ocfs2_pack_timespec(&inode->i_atime));
2187 lvb->lvb_ictime_packed =
2188 cpu_to_be64(ocfs2_pack_timespec(&inode->i_ctime));
2189 lvb->lvb_imtime_packed =
2190 cpu_to_be64(ocfs2_pack_timespec(&inode->i_mtime));
2191 lvb->lvb_iattr = cpu_to_be32(oi->ip_attr);
2192 lvb->lvb_idynfeatures = cpu_to_be16(oi->ip_dyn_features);
2193 lvb->lvb_igeneration = cpu_to_be32(inode->i_generation);
2196 mlog_meta_lvb(0, lockres);
2199 static void ocfs2_unpack_timespec(struct timespec64 *spec,
2202 spec->tv_sec = packed_time >> OCFS2_SEC_SHIFT;
2203 spec->tv_nsec = packed_time & OCFS2_NSEC_MASK;
2206 static int ocfs2_refresh_inode_from_lvb(struct inode *inode)
2208 struct ocfs2_inode_info *oi = OCFS2_I(inode);
2209 struct ocfs2_lock_res *lockres = &oi->ip_inode_lockres;
2210 struct ocfs2_meta_lvb *lvb;
2212 mlog_meta_lvb(0, lockres);
2214 lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2215 if (inode_wrong_type(inode, be16_to_cpu(lvb->lvb_imode)))
2218 /* We're safe here without the lockres lock... */
2219 spin_lock(&oi->ip_lock);
2220 oi->ip_clusters = be32_to_cpu(lvb->lvb_iclusters);
2221 i_size_write(inode, be64_to_cpu(lvb->lvb_isize));
2223 oi->ip_attr = be32_to_cpu(lvb->lvb_iattr);
2224 oi->ip_dyn_features = be16_to_cpu(lvb->lvb_idynfeatures);
2225 ocfs2_set_inode_flags(inode);
2227 /* fast-symlinks are a special case */
2228 if (S_ISLNK(inode->i_mode) && !oi->ip_clusters)
2229 inode->i_blocks = 0;
2231 inode->i_blocks = ocfs2_inode_sector_count(inode);
2233 i_uid_write(inode, be32_to_cpu(lvb->lvb_iuid));
2234 i_gid_write(inode, be32_to_cpu(lvb->lvb_igid));
2235 inode->i_mode = be16_to_cpu(lvb->lvb_imode);
2236 set_nlink(inode, be16_to_cpu(lvb->lvb_inlink));
2237 ocfs2_unpack_timespec(&inode->i_atime,
2238 be64_to_cpu(lvb->lvb_iatime_packed));
2239 ocfs2_unpack_timespec(&inode->i_mtime,
2240 be64_to_cpu(lvb->lvb_imtime_packed));
2241 ocfs2_unpack_timespec(&inode->i_ctime,
2242 be64_to_cpu(lvb->lvb_ictime_packed));
2243 spin_unlock(&oi->ip_lock);
2247 static inline int ocfs2_meta_lvb_is_trustable(struct inode *inode,
2248 struct ocfs2_lock_res *lockres)
2250 struct ocfs2_meta_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2252 if (ocfs2_dlm_lvb_valid(&lockres->l_lksb)
2253 && lvb->lvb_version == OCFS2_LVB_VERSION
2254 && be32_to_cpu(lvb->lvb_igeneration) == inode->i_generation)
2259 /* Determine whether a lock resource needs to be refreshed, and
2260 * arbitrate who gets to refresh it.
2262 * 0 means no refresh needed.
2264 * > 0 means you need to refresh this and you MUST call
2265 * ocfs2_complete_lock_res_refresh afterwards. */
2266 static int ocfs2_should_refresh_lock_res(struct ocfs2_lock_res *lockres)
2268 unsigned long flags;
2272 spin_lock_irqsave(&lockres->l_lock, flags);
2273 if (!(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH)) {
2274 spin_unlock_irqrestore(&lockres->l_lock, flags);
2278 if (lockres->l_flags & OCFS2_LOCK_REFRESHING) {
2279 spin_unlock_irqrestore(&lockres->l_lock, flags);
2281 ocfs2_wait_on_refreshing_lock(lockres);
2285 /* Ok, I'll be the one to refresh this lock. */
2286 lockres_or_flags(lockres, OCFS2_LOCK_REFRESHING);
2287 spin_unlock_irqrestore(&lockres->l_lock, flags);
2291 mlog(0, "status %d\n", status);
2295 /* If status is non zero, I'll mark it as not being in refresh
2296 * anymroe, but i won't clear the needs refresh flag. */
2297 static inline void ocfs2_complete_lock_res_refresh(struct ocfs2_lock_res *lockres,
2300 unsigned long flags;
2302 spin_lock_irqsave(&lockres->l_lock, flags);
2303 lockres_clear_flags(lockres, OCFS2_LOCK_REFRESHING);
2305 lockres_clear_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
2306 spin_unlock_irqrestore(&lockres->l_lock, flags);
2308 wake_up(&lockres->l_event);
2311 /* may or may not return a bh if it went to disk. */
2312 static int ocfs2_inode_lock_update(struct inode *inode,
2313 struct buffer_head **bh)
2316 struct ocfs2_inode_info *oi = OCFS2_I(inode);
2317 struct ocfs2_lock_res *lockres = &oi->ip_inode_lockres;
2318 struct ocfs2_dinode *fe;
2319 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
2321 if (ocfs2_mount_local(osb))
2324 spin_lock(&oi->ip_lock);
2325 if (oi->ip_flags & OCFS2_INODE_DELETED) {
2326 mlog(0, "Orphaned inode %llu was deleted while we "
2327 "were waiting on a lock. ip_flags = 0x%x\n",
2328 (unsigned long long)oi->ip_blkno, oi->ip_flags);
2329 spin_unlock(&oi->ip_lock);
2333 spin_unlock(&oi->ip_lock);
2335 if (!ocfs2_should_refresh_lock_res(lockres))
2338 /* This will discard any caching information we might have had
2339 * for the inode metadata. */
2340 ocfs2_metadata_cache_purge(INODE_CACHE(inode));
2342 ocfs2_extent_map_trunc(inode, 0);
2344 if (ocfs2_meta_lvb_is_trustable(inode, lockres)) {
2345 mlog(0, "Trusting LVB on inode %llu\n",
2346 (unsigned long long)oi->ip_blkno);
2347 status = ocfs2_refresh_inode_from_lvb(inode);
2350 /* Boo, we have to go to disk. */
2351 /* read bh, cast, ocfs2_refresh_inode */
2352 status = ocfs2_read_inode_block(inode, bh);
2357 fe = (struct ocfs2_dinode *) (*bh)->b_data;
2358 if (inode_wrong_type(inode, le16_to_cpu(fe->i_mode))) {
2363 /* This is a good chance to make sure we're not
2364 * locking an invalid object. ocfs2_read_inode_block()
2365 * already checked that the inode block is sane.
2367 * We bug on a stale inode here because we checked
2368 * above whether it was wiped from disk. The wiping
2369 * node provides a guarantee that we receive that
2370 * message and can mark the inode before dropping any
2371 * locks associated with it. */
2372 mlog_bug_on_msg(inode->i_generation !=
2373 le32_to_cpu(fe->i_generation),
2374 "Invalid dinode %llu disk generation: %u "
2375 "inode->i_generation: %u\n",
2376 (unsigned long long)oi->ip_blkno,
2377 le32_to_cpu(fe->i_generation),
2378 inode->i_generation);
2379 mlog_bug_on_msg(le64_to_cpu(fe->i_dtime) ||
2380 !(fe->i_flags & cpu_to_le32(OCFS2_VALID_FL)),
2381 "Stale dinode %llu dtime: %llu flags: 0x%x\n",
2382 (unsigned long long)oi->ip_blkno,
2383 (unsigned long long)le64_to_cpu(fe->i_dtime),
2384 le32_to_cpu(fe->i_flags));
2386 ocfs2_refresh_inode(inode, fe);
2387 ocfs2_track_lock_refresh(lockres);
2392 ocfs2_complete_lock_res_refresh(lockres, status);
2397 static int ocfs2_assign_bh(struct inode *inode,
2398 struct buffer_head **ret_bh,
2399 struct buffer_head *passed_bh)
2404 /* Ok, the update went to disk for us, use the
2406 *ret_bh = passed_bh;
2412 status = ocfs2_read_inode_block(inode, ret_bh);
2420 * returns < 0 error if the callback will never be called, otherwise
2421 * the result of the lock will be communicated via the callback.
2423 int ocfs2_inode_lock_full_nested(struct inode *inode,
2424 struct buffer_head **ret_bh,
2429 int status, level, acquired;
2431 struct ocfs2_lock_res *lockres = NULL;
2432 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
2433 struct buffer_head *local_bh = NULL;
2435 mlog(0, "inode %llu, take %s META lock\n",
2436 (unsigned long long)OCFS2_I(inode)->ip_blkno,
2437 ex ? "EXMODE" : "PRMODE");
2441 /* We'll allow faking a readonly metadata lock for
2443 if (ocfs2_is_hard_readonly(osb)) {
2449 if ((arg_flags & OCFS2_META_LOCK_GETBH) ||
2450 ocfs2_mount_local(osb))
2453 if (!(arg_flags & OCFS2_META_LOCK_RECOVERY))
2454 ocfs2_wait_for_recovery(osb);
2456 lockres = &OCFS2_I(inode)->ip_inode_lockres;
2457 level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2459 if (arg_flags & OCFS2_META_LOCK_NOQUEUE)
2460 dlm_flags |= DLM_LKF_NOQUEUE;
2462 status = __ocfs2_cluster_lock(osb, lockres, level, dlm_flags,
2463 arg_flags, subclass, _RET_IP_);
2465 if (status != -EAGAIN)
2470 /* Notify the error cleanup path to drop the cluster lock. */
2473 /* We wait twice because a node may have died while we were in
2474 * the lower dlm layers. The second time though, we've
2475 * committed to owning this lock so we don't allow signals to
2476 * abort the operation. */
2477 if (!(arg_flags & OCFS2_META_LOCK_RECOVERY))
2478 ocfs2_wait_for_recovery(osb);
2482 * We only see this flag if we're being called from
2483 * ocfs2_read_locked_inode(). It means we're locking an inode
2484 * which hasn't been populated yet, so clear the refresh flag
2485 * and let the caller handle it.
2487 if (inode->i_state & I_NEW) {
2490 ocfs2_complete_lock_res_refresh(lockres, 0);
2494 /* This is fun. The caller may want a bh back, or it may
2495 * not. ocfs2_inode_lock_update definitely wants one in, but
2496 * may or may not read one, depending on what's in the
2497 * LVB. The result of all of this is that we've *only* gone to
2498 * disk if we have to, so the complexity is worthwhile. */
2499 status = ocfs2_inode_lock_update(inode, &local_bh);
2501 if (status != -ENOENT)
2507 status = ocfs2_assign_bh(inode, ret_bh, local_bh);
2516 if (ret_bh && (*ret_bh)) {
2521 ocfs2_inode_unlock(inode, ex);
2529 * This is working around a lock inversion between tasks acquiring DLM
2530 * locks while holding a page lock and the downconvert thread which
2531 * blocks dlm lock acquiry while acquiring page locks.
2533 * ** These _with_page variantes are only intended to be called from aop
2534 * methods that hold page locks and return a very specific *positive* error
2535 * code that aop methods pass up to the VFS -- test for errors with != 0. **
2537 * The DLM is called such that it returns -EAGAIN if it would have
2538 * blocked waiting for the downconvert thread. In that case we unlock
2539 * our page so the downconvert thread can make progress. Once we've
2540 * done this we have to return AOP_TRUNCATED_PAGE so the aop method
2541 * that called us can bubble that back up into the VFS who will then
2542 * immediately retry the aop call.
2544 int ocfs2_inode_lock_with_page(struct inode *inode,
2545 struct buffer_head **ret_bh,
2551 ret = ocfs2_inode_lock_full(inode, ret_bh, ex, OCFS2_LOCK_NONBLOCK);
2552 if (ret == -EAGAIN) {
2555 * If we can't get inode lock immediately, we should not return
2556 * directly here, since this will lead to a softlockup problem.
2557 * The method is to get a blocking lock and immediately unlock
2558 * before returning, this can avoid CPU resource waste due to
2559 * lots of retries, and benefits fairness in getting lock.
2561 if (ocfs2_inode_lock(inode, ret_bh, ex) == 0)
2562 ocfs2_inode_unlock(inode, ex);
2563 ret = AOP_TRUNCATED_PAGE;
2569 int ocfs2_inode_lock_atime(struct inode *inode,
2570 struct vfsmount *vfsmnt,
2571 int *level, int wait)
2576 ret = ocfs2_inode_lock(inode, NULL, 0);
2578 ret = ocfs2_try_inode_lock(inode, NULL, 0);
2587 * If we should update atime, we will get EX lock,
2588 * otherwise we just get PR lock.
2590 if (ocfs2_should_update_atime(inode, vfsmnt)) {
2591 struct buffer_head *bh = NULL;
2593 ocfs2_inode_unlock(inode, 0);
2595 ret = ocfs2_inode_lock(inode, &bh, 1);
2597 ret = ocfs2_try_inode_lock(inode, &bh, 1);
2605 if (ocfs2_should_update_atime(inode, vfsmnt))
2606 ocfs2_update_inode_atime(inode, bh);
2614 void ocfs2_inode_unlock(struct inode *inode,
2617 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2618 struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_inode_lockres;
2619 struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
2621 mlog(0, "inode %llu drop %s META lock\n",
2622 (unsigned long long)OCFS2_I(inode)->ip_blkno,
2623 ex ? "EXMODE" : "PRMODE");
2625 if (!ocfs2_is_hard_readonly(osb) &&
2626 !ocfs2_mount_local(osb))
2627 ocfs2_cluster_unlock(osb, lockres, level);
2631 * This _tracker variantes are introduced to deal with the recursive cluster
2632 * locking issue. The idea is to keep track of a lock holder on the stack of
2633 * the current process. If there's a lock holder on the stack, we know the
2634 * task context is already protected by cluster locking. Currently, they're
2635 * used in some VFS entry routines.
2637 * return < 0 on error, return == 0 if there's no lock holder on the stack
2638 * before this call, return == 1 if this call would be a recursive locking.
2639 * return == -1 if this lock attempt will cause an upgrade which is forbidden.
2641 * When taking lock levels into account,we face some different situations.
2643 * 1. no lock is held
2644 * In this case, just lock the inode as requested and return 0
2646 * 2. We are holding a lock
2647 * For this situation, things diverges into several cases
2649 * wanted holding what to do
2650 * ex ex see 2.1 below
2651 * ex pr see 2.2 below
2652 * pr ex see 2.1 below
2653 * pr pr see 2.1 below
2655 * 2.1 lock level that is been held is compatible
2656 * with the wanted level, so no lock action will be tacken.
2658 * 2.2 Otherwise, an upgrade is needed, but it is forbidden.
2660 * Reason why upgrade within a process is forbidden is that
2661 * lock upgrade may cause dead lock. The following illustrates
2664 * thread on node1 thread on node2
2665 * ocfs2_inode_lock_tracker(ex=0)
2667 * <====== ocfs2_inode_lock_tracker(ex=1)
2669 * ocfs2_inode_lock_tracker(ex=1)
2671 int ocfs2_inode_lock_tracker(struct inode *inode,
2672 struct buffer_head **ret_bh,
2674 struct ocfs2_lock_holder *oh)
2677 struct ocfs2_lock_res *lockres;
2678 struct ocfs2_lock_holder *tmp_oh;
2679 struct pid *pid = task_pid(current);
2682 lockres = &OCFS2_I(inode)->ip_inode_lockres;
2683 tmp_oh = ocfs2_pid_holder(lockres, pid);
2687 * This corresponds to the case 1.
2688 * We haven't got any lock before.
2690 status = ocfs2_inode_lock_full(inode, ret_bh, ex, 0);
2692 if (status != -ENOENT)
2698 ocfs2_add_holder(lockres, oh);
2702 if (unlikely(ex && !tmp_oh->oh_ex)) {
2704 * case 2.2 upgrade may cause dead lock, forbid it.
2706 mlog(ML_ERROR, "Recursive locking is not permitted to "
2707 "upgrade to EX level from PR level.\n");
2713 * case 2.1 OCFS2_META_LOCK_GETBH flag make ocfs2_inode_lock_full.
2714 * ignore the lock level and just update it.
2717 status = ocfs2_inode_lock_full(inode, ret_bh, ex,
2718 OCFS2_META_LOCK_GETBH);
2720 if (status != -ENOENT)
2728 void ocfs2_inode_unlock_tracker(struct inode *inode,
2730 struct ocfs2_lock_holder *oh,
2733 struct ocfs2_lock_res *lockres;
2735 lockres = &OCFS2_I(inode)->ip_inode_lockres;
2736 /* had_lock means that the currect process already takes the cluster
2738 * If had_lock is 1, we have nothing to do here.
2739 * If had_lock is 0, we will release the lock.
2742 ocfs2_inode_unlock(inode, oh->oh_ex);
2743 ocfs2_remove_holder(lockres, oh);
2747 int ocfs2_orphan_scan_lock(struct ocfs2_super *osb, u32 *seqno)
2749 struct ocfs2_lock_res *lockres;
2750 struct ocfs2_orphan_scan_lvb *lvb;
2753 if (ocfs2_is_hard_readonly(osb))
2756 if (ocfs2_mount_local(osb))
2759 lockres = &osb->osb_orphan_scan.os_lockres;
2760 status = ocfs2_cluster_lock(osb, lockres, DLM_LOCK_EX, 0, 0);
2764 lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2765 if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
2766 lvb->lvb_version == OCFS2_ORPHAN_LVB_VERSION)
2767 *seqno = be32_to_cpu(lvb->lvb_os_seqno);
2769 *seqno = osb->osb_orphan_scan.os_seqno + 1;
2774 void ocfs2_orphan_scan_unlock(struct ocfs2_super *osb, u32 seqno)
2776 struct ocfs2_lock_res *lockres;
2777 struct ocfs2_orphan_scan_lvb *lvb;
2779 if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb)) {
2780 lockres = &osb->osb_orphan_scan.os_lockres;
2781 lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2782 lvb->lvb_version = OCFS2_ORPHAN_LVB_VERSION;
2783 lvb->lvb_os_seqno = cpu_to_be32(seqno);
2784 ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
2788 int ocfs2_super_lock(struct ocfs2_super *osb,
2792 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2793 struct ocfs2_lock_res *lockres = &osb->osb_super_lockres;
2795 if (ocfs2_is_hard_readonly(osb))
2798 if (ocfs2_mount_local(osb))
2801 status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
2807 /* The super block lock path is really in the best position to
2808 * know when resources covered by the lock need to be
2809 * refreshed, so we do it here. Of course, making sense of
2810 * everything is up to the caller :) */
2811 status = ocfs2_should_refresh_lock_res(lockres);
2813 status = ocfs2_refresh_slot_info(osb);
2815 ocfs2_complete_lock_res_refresh(lockres, status);
2818 ocfs2_cluster_unlock(osb, lockres, level);
2821 ocfs2_track_lock_refresh(lockres);
2827 void ocfs2_super_unlock(struct ocfs2_super *osb,
2830 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2831 struct ocfs2_lock_res *lockres = &osb->osb_super_lockres;
2833 if (!ocfs2_mount_local(osb))
2834 ocfs2_cluster_unlock(osb, lockres, level);
2837 int ocfs2_rename_lock(struct ocfs2_super *osb)
2840 struct ocfs2_lock_res *lockres = &osb->osb_rename_lockres;
2842 if (ocfs2_is_hard_readonly(osb))
2845 if (ocfs2_mount_local(osb))
2848 status = ocfs2_cluster_lock(osb, lockres, DLM_LOCK_EX, 0, 0);
2855 void ocfs2_rename_unlock(struct ocfs2_super *osb)
2857 struct ocfs2_lock_res *lockres = &osb->osb_rename_lockres;
2859 if (!ocfs2_mount_local(osb))
2860 ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
2863 int ocfs2_nfs_sync_lock(struct ocfs2_super *osb, int ex)
2866 struct ocfs2_lock_res *lockres = &osb->osb_nfs_sync_lockres;
2868 if (ocfs2_is_hard_readonly(osb))
2872 down_write(&osb->nfs_sync_rwlock);
2874 down_read(&osb->nfs_sync_rwlock);
2876 if (ocfs2_mount_local(osb))
2879 status = ocfs2_cluster_lock(osb, lockres, ex ? LKM_EXMODE : LKM_PRMODE,
2882 mlog(ML_ERROR, "lock on nfs sync lock failed %d\n", status);
2885 up_write(&osb->nfs_sync_rwlock);
2887 up_read(&osb->nfs_sync_rwlock);
2893 void ocfs2_nfs_sync_unlock(struct ocfs2_super *osb, int ex)
2895 struct ocfs2_lock_res *lockres = &osb->osb_nfs_sync_lockres;
2897 if (!ocfs2_mount_local(osb))
2898 ocfs2_cluster_unlock(osb, lockres,
2899 ex ? LKM_EXMODE : LKM_PRMODE);
2901 up_write(&osb->nfs_sync_rwlock);
2903 up_read(&osb->nfs_sync_rwlock);
2906 int ocfs2_trim_fs_lock(struct ocfs2_super *osb,
2907 struct ocfs2_trim_fs_info *info, int trylock)
2910 struct ocfs2_trim_fs_lvb *lvb;
2911 struct ocfs2_lock_res *lockres = &osb->osb_trim_fs_lockres;
2916 if (ocfs2_is_hard_readonly(osb))
2919 if (ocfs2_mount_local(osb))
2922 status = ocfs2_cluster_lock(osb, lockres, DLM_LOCK_EX,
2923 trylock ? DLM_LKF_NOQUEUE : 0, 0);
2925 if (status != -EAGAIN)
2931 lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2932 if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
2933 lvb->lvb_version == OCFS2_TRIMFS_LVB_VERSION) {
2935 info->tf_success = lvb->lvb_success;
2936 info->tf_nodenum = be32_to_cpu(lvb->lvb_nodenum);
2937 info->tf_start = be64_to_cpu(lvb->lvb_start);
2938 info->tf_len = be64_to_cpu(lvb->lvb_len);
2939 info->tf_minlen = be64_to_cpu(lvb->lvb_minlen);
2940 info->tf_trimlen = be64_to_cpu(lvb->lvb_trimlen);
2947 void ocfs2_trim_fs_unlock(struct ocfs2_super *osb,
2948 struct ocfs2_trim_fs_info *info)
2950 struct ocfs2_trim_fs_lvb *lvb;
2951 struct ocfs2_lock_res *lockres = &osb->osb_trim_fs_lockres;
2953 if (ocfs2_mount_local(osb))
2957 lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2958 lvb->lvb_version = OCFS2_TRIMFS_LVB_VERSION;
2959 lvb->lvb_success = info->tf_success;
2960 lvb->lvb_nodenum = cpu_to_be32(info->tf_nodenum);
2961 lvb->lvb_start = cpu_to_be64(info->tf_start);
2962 lvb->lvb_len = cpu_to_be64(info->tf_len);
2963 lvb->lvb_minlen = cpu_to_be64(info->tf_minlen);
2964 lvb->lvb_trimlen = cpu_to_be64(info->tf_trimlen);
2967 ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
2970 int ocfs2_dentry_lock(struct dentry *dentry, int ex)
2973 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2974 struct ocfs2_dentry_lock *dl = dentry->d_fsdata;
2975 struct ocfs2_super *osb = OCFS2_SB(dentry->d_sb);
2979 if (ocfs2_is_hard_readonly(osb)) {
2985 if (ocfs2_mount_local(osb))
2988 ret = ocfs2_cluster_lock(osb, &dl->dl_lockres, level, 0, 0);
2995 void ocfs2_dentry_unlock(struct dentry *dentry, int ex)
2997 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2998 struct ocfs2_dentry_lock *dl = dentry->d_fsdata;
2999 struct ocfs2_super *osb = OCFS2_SB(dentry->d_sb);
3001 if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb))
3002 ocfs2_cluster_unlock(osb, &dl->dl_lockres, level);
3005 /* Reference counting of the dlm debug structure. We want this because
3006 * open references on the debug inodes can live on after a mount, so
3007 * we can't rely on the ocfs2_super to always exist. */
3008 static void ocfs2_dlm_debug_free(struct kref *kref)
3010 struct ocfs2_dlm_debug *dlm_debug;
3012 dlm_debug = container_of(kref, struct ocfs2_dlm_debug, d_refcnt);
3017 void ocfs2_put_dlm_debug(struct ocfs2_dlm_debug *dlm_debug)
3020 kref_put(&dlm_debug->d_refcnt, ocfs2_dlm_debug_free);
3023 static void ocfs2_get_dlm_debug(struct ocfs2_dlm_debug *debug)
3025 kref_get(&debug->d_refcnt);
3028 struct ocfs2_dlm_debug *ocfs2_new_dlm_debug(void)
3030 struct ocfs2_dlm_debug *dlm_debug;
3032 dlm_debug = kmalloc(sizeof(struct ocfs2_dlm_debug), GFP_KERNEL);
3034 mlog_errno(-ENOMEM);
3038 kref_init(&dlm_debug->d_refcnt);
3039 INIT_LIST_HEAD(&dlm_debug->d_lockres_tracking);
3040 dlm_debug->d_filter_secs = 0;
3045 /* Access to this is arbitrated for us via seq_file->sem. */
3046 struct ocfs2_dlm_seq_priv {
3047 struct ocfs2_dlm_debug *p_dlm_debug;
3048 struct ocfs2_lock_res p_iter_res;
3049 struct ocfs2_lock_res p_tmp_res;
3052 static struct ocfs2_lock_res *ocfs2_dlm_next_res(struct ocfs2_lock_res *start,
3053 struct ocfs2_dlm_seq_priv *priv)
3055 struct ocfs2_lock_res *iter, *ret = NULL;
3056 struct ocfs2_dlm_debug *dlm_debug = priv->p_dlm_debug;
3058 assert_spin_locked(&ocfs2_dlm_tracking_lock);
3060 list_for_each_entry(iter, &start->l_debug_list, l_debug_list) {
3061 /* discover the head of the list */
3062 if (&iter->l_debug_list == &dlm_debug->d_lockres_tracking) {
3063 mlog(0, "End of list found, %p\n", ret);
3067 /* We track our "dummy" iteration lockres' by a NULL
3069 if (iter->l_ops != NULL) {
3078 static void *ocfs2_dlm_seq_start(struct seq_file *m, loff_t *pos)
3080 struct ocfs2_dlm_seq_priv *priv = m->private;
3081 struct ocfs2_lock_res *iter;
3083 spin_lock(&ocfs2_dlm_tracking_lock);
3084 iter = ocfs2_dlm_next_res(&priv->p_iter_res, priv);
3086 /* Since lockres' have the lifetime of their container
3087 * (which can be inodes, ocfs2_supers, etc) we want to
3088 * copy this out to a temporary lockres while still
3089 * under the spinlock. Obviously after this we can't
3090 * trust any pointers on the copy returned, but that's
3091 * ok as the information we want isn't typically held
3093 priv->p_tmp_res = *iter;
3094 iter = &priv->p_tmp_res;
3096 spin_unlock(&ocfs2_dlm_tracking_lock);
3101 static void ocfs2_dlm_seq_stop(struct seq_file *m, void *v)
3105 static void *ocfs2_dlm_seq_next(struct seq_file *m, void *v, loff_t *pos)
3107 struct ocfs2_dlm_seq_priv *priv = m->private;
3108 struct ocfs2_lock_res *iter = v;
3109 struct ocfs2_lock_res *dummy = &priv->p_iter_res;
3111 spin_lock(&ocfs2_dlm_tracking_lock);
3112 iter = ocfs2_dlm_next_res(iter, priv);
3113 list_del_init(&dummy->l_debug_list);
3115 list_add(&dummy->l_debug_list, &iter->l_debug_list);
3116 priv->p_tmp_res = *iter;
3117 iter = &priv->p_tmp_res;
3119 spin_unlock(&ocfs2_dlm_tracking_lock);
3125 * Version is used by debugfs.ocfs2 to determine the format being used
3128 * - Lock stats printed
3130 * - Max time in lock stats is in usecs (instead of nsecs)
3132 * - Add last pr/ex unlock times and first lock wait time in usecs
3134 #define OCFS2_DLM_DEBUG_STR_VERSION 4
3135 static int ocfs2_dlm_seq_show(struct seq_file *m, void *v)
3139 struct ocfs2_lock_res *lockres = v;
3140 #ifdef CONFIG_OCFS2_FS_STATS
3142 struct ocfs2_dlm_debug *dlm_debug =
3143 ((struct ocfs2_dlm_seq_priv *)m->private)->p_dlm_debug;
3149 #ifdef CONFIG_OCFS2_FS_STATS
3150 if (!lockres->l_lock_wait && dlm_debug->d_filter_secs) {
3151 now = ktime_to_us(ktime_get_real());
3152 if (lockres->l_lock_prmode.ls_last >
3153 lockres->l_lock_exmode.ls_last)
3154 last = lockres->l_lock_prmode.ls_last;
3156 last = lockres->l_lock_exmode.ls_last;
3158 * Use d_filter_secs field to filter lock resources dump,
3159 * the default d_filter_secs(0) value filters nothing,
3160 * otherwise, only dump the last N seconds active lock
3163 if (div_u64(now - last, 1000000) > dlm_debug->d_filter_secs)
3168 seq_printf(m, "0x%x\t", OCFS2_DLM_DEBUG_STR_VERSION);
3170 if (lockres->l_type == OCFS2_LOCK_TYPE_DENTRY)
3171 seq_printf(m, "%.*s%08x\t", OCFS2_DENTRY_LOCK_INO_START - 1,
3173 (unsigned int)ocfs2_get_dentry_lock_ino(lockres));
3175 seq_printf(m, "%.*s\t", OCFS2_LOCK_ID_MAX_LEN, lockres->l_name);
3177 seq_printf(m, "%d\t"
3188 lockres->l_unlock_action,
3189 lockres->l_ro_holders,
3190 lockres->l_ex_holders,
3191 lockres->l_requested,
3192 lockres->l_blocking);
3194 /* Dump the raw LVB */
3195 lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
3196 for(i = 0; i < DLM_LVB_LEN; i++)
3197 seq_printf(m, "0x%x\t", lvb[i]);
3199 #ifdef CONFIG_OCFS2_FS_STATS
3200 # define lock_num_prmode(_l) ((_l)->l_lock_prmode.ls_gets)
3201 # define lock_num_exmode(_l) ((_l)->l_lock_exmode.ls_gets)
3202 # define lock_num_prmode_failed(_l) ((_l)->l_lock_prmode.ls_fail)
3203 # define lock_num_exmode_failed(_l) ((_l)->l_lock_exmode.ls_fail)
3204 # define lock_total_prmode(_l) ((_l)->l_lock_prmode.ls_total)
3205 # define lock_total_exmode(_l) ((_l)->l_lock_exmode.ls_total)
3206 # define lock_max_prmode(_l) ((_l)->l_lock_prmode.ls_max)
3207 # define lock_max_exmode(_l) ((_l)->l_lock_exmode.ls_max)
3208 # define lock_refresh(_l) ((_l)->l_lock_refresh)
3209 # define lock_last_prmode(_l) ((_l)->l_lock_prmode.ls_last)
3210 # define lock_last_exmode(_l) ((_l)->l_lock_exmode.ls_last)
3211 # define lock_wait(_l) ((_l)->l_lock_wait)
3213 # define lock_num_prmode(_l) (0)
3214 # define lock_num_exmode(_l) (0)
3215 # define lock_num_prmode_failed(_l) (0)
3216 # define lock_num_exmode_failed(_l) (0)
3217 # define lock_total_prmode(_l) (0ULL)
3218 # define lock_total_exmode(_l) (0ULL)
3219 # define lock_max_prmode(_l) (0)
3220 # define lock_max_exmode(_l) (0)
3221 # define lock_refresh(_l) (0)
3222 # define lock_last_prmode(_l) (0ULL)
3223 # define lock_last_exmode(_l) (0ULL)
3224 # define lock_wait(_l) (0ULL)
3226 /* The following seq_print was added in version 2 of this output */
3227 seq_printf(m, "%u\t"
3239 lock_num_prmode(lockres),
3240 lock_num_exmode(lockres),
3241 lock_num_prmode_failed(lockres),
3242 lock_num_exmode_failed(lockres),
3243 lock_total_prmode(lockres),
3244 lock_total_exmode(lockres),
3245 lock_max_prmode(lockres),
3246 lock_max_exmode(lockres),
3247 lock_refresh(lockres),
3248 lock_last_prmode(lockres),
3249 lock_last_exmode(lockres),
3250 lock_wait(lockres));
3253 seq_printf(m, "\n");
3257 static const struct seq_operations ocfs2_dlm_seq_ops = {
3258 .start = ocfs2_dlm_seq_start,
3259 .stop = ocfs2_dlm_seq_stop,
3260 .next = ocfs2_dlm_seq_next,
3261 .show = ocfs2_dlm_seq_show,
3264 static int ocfs2_dlm_debug_release(struct inode *inode, struct file *file)
3266 struct seq_file *seq = file->private_data;
3267 struct ocfs2_dlm_seq_priv *priv = seq->private;
3268 struct ocfs2_lock_res *res = &priv->p_iter_res;
3270 ocfs2_remove_lockres_tracking(res);
3271 ocfs2_put_dlm_debug(priv->p_dlm_debug);
3272 return seq_release_private(inode, file);
3275 static int ocfs2_dlm_debug_open(struct inode *inode, struct file *file)
3277 struct ocfs2_dlm_seq_priv *priv;
3278 struct ocfs2_super *osb;
3280 priv = __seq_open_private(file, &ocfs2_dlm_seq_ops, sizeof(*priv));
3282 mlog_errno(-ENOMEM);
3286 osb = inode->i_private;
3287 ocfs2_get_dlm_debug(osb->osb_dlm_debug);
3288 priv->p_dlm_debug = osb->osb_dlm_debug;
3289 INIT_LIST_HEAD(&priv->p_iter_res.l_debug_list);
3291 ocfs2_add_lockres_tracking(&priv->p_iter_res,
3297 static const struct file_operations ocfs2_dlm_debug_fops = {
3298 .open = ocfs2_dlm_debug_open,
3299 .release = ocfs2_dlm_debug_release,
3301 .llseek = seq_lseek,
3304 static void ocfs2_dlm_init_debug(struct ocfs2_super *osb)
3306 struct ocfs2_dlm_debug *dlm_debug = osb->osb_dlm_debug;
3308 debugfs_create_file("locking_state", S_IFREG|S_IRUSR,
3309 osb->osb_debug_root, osb, &ocfs2_dlm_debug_fops);
3311 debugfs_create_u32("locking_filter", 0600, osb->osb_debug_root,
3312 &dlm_debug->d_filter_secs);
3313 ocfs2_get_dlm_debug(dlm_debug);
3316 static void ocfs2_dlm_shutdown_debug(struct ocfs2_super *osb)
3318 struct ocfs2_dlm_debug *dlm_debug = osb->osb_dlm_debug;
3321 ocfs2_put_dlm_debug(dlm_debug);
3324 int ocfs2_dlm_init(struct ocfs2_super *osb)
3327 struct ocfs2_cluster_connection *conn = NULL;
3329 if (ocfs2_mount_local(osb)) {
3334 ocfs2_dlm_init_debug(osb);
3336 /* launch downconvert thread */
3337 osb->dc_task = kthread_run(ocfs2_downconvert_thread, osb, "ocfs2dc-%s",
3339 if (IS_ERR(osb->dc_task)) {
3340 status = PTR_ERR(osb->dc_task);
3341 osb->dc_task = NULL;
3346 /* for now, uuid == domain */
3347 status = ocfs2_cluster_connect(osb->osb_cluster_stack,
3348 osb->osb_cluster_name,
3349 strlen(osb->osb_cluster_name),
3351 strlen(osb->uuid_str),
3352 &lproto, ocfs2_do_node_down, osb,
3359 status = ocfs2_cluster_this_node(conn, &osb->node_num);
3363 "could not find this host's node number\n");
3364 ocfs2_cluster_disconnect(conn, 0);
3369 ocfs2_super_lock_res_init(&osb->osb_super_lockres, osb);
3370 ocfs2_rename_lock_res_init(&osb->osb_rename_lockres, osb);
3371 ocfs2_nfs_sync_lock_init(osb);
3372 ocfs2_orphan_scan_lock_res_init(&osb->osb_orphan_scan.os_lockres, osb);
3377 ocfs2_dlm_shutdown_debug(osb);
3379 kthread_stop(osb->dc_task);
3385 void ocfs2_dlm_shutdown(struct ocfs2_super *osb,
3388 ocfs2_drop_osb_locks(osb);
3391 * Now that we have dropped all locks and ocfs2_dismount_volume()
3392 * has disabled recovery, the DLM won't be talking to us. It's
3393 * safe to tear things down before disconnecting the cluster.
3397 kthread_stop(osb->dc_task);
3398 osb->dc_task = NULL;
3401 ocfs2_lock_res_free(&osb->osb_super_lockres);
3402 ocfs2_lock_res_free(&osb->osb_rename_lockres);
3403 ocfs2_lock_res_free(&osb->osb_nfs_sync_lockres);
3404 ocfs2_lock_res_free(&osb->osb_orphan_scan.os_lockres);
3406 ocfs2_cluster_disconnect(osb->cconn, hangup_pending);
3409 ocfs2_dlm_shutdown_debug(osb);
3412 static int ocfs2_drop_lock(struct ocfs2_super *osb,
3413 struct ocfs2_lock_res *lockres)
3416 unsigned long flags;
3419 /* We didn't get anywhere near actually using this lockres. */
3420 if (!(lockres->l_flags & OCFS2_LOCK_INITIALIZED))
3423 if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
3424 lkm_flags |= DLM_LKF_VALBLK;
3426 spin_lock_irqsave(&lockres->l_lock, flags);
3428 mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_FREEING),
3429 "lockres %s, flags 0x%lx\n",
3430 lockres->l_name, lockres->l_flags);
3432 while (lockres->l_flags & OCFS2_LOCK_BUSY) {
3433 mlog(0, "waiting on busy lock \"%s\": flags = %lx, action = "
3434 "%u, unlock_action = %u\n",
3435 lockres->l_name, lockres->l_flags, lockres->l_action,
3436 lockres->l_unlock_action);
3438 spin_unlock_irqrestore(&lockres->l_lock, flags);
3440 /* XXX: Today we just wait on any busy
3441 * locks... Perhaps we need to cancel converts in the
3443 ocfs2_wait_on_busy_lock(lockres);
3445 spin_lock_irqsave(&lockres->l_lock, flags);
3448 if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
3449 if (lockres->l_flags & OCFS2_LOCK_ATTACHED &&
3450 lockres->l_level == DLM_LOCK_EX &&
3451 !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
3452 lockres->l_ops->set_lvb(lockres);
3455 if (lockres->l_flags & OCFS2_LOCK_BUSY)
3456 mlog(ML_ERROR, "destroying busy lock: \"%s\"\n",
3458 if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
3459 mlog(0, "destroying blocked lock: \"%s\"\n", lockres->l_name);
3461 if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
3462 spin_unlock_irqrestore(&lockres->l_lock, flags);
3466 lockres_clear_flags(lockres, OCFS2_LOCK_ATTACHED);
3468 /* make sure we never get here while waiting for an ast to
3470 BUG_ON(lockres->l_action != OCFS2_AST_INVALID);
3472 /* is this necessary? */
3473 lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
3474 lockres->l_unlock_action = OCFS2_UNLOCK_DROP_LOCK;
3475 spin_unlock_irqrestore(&lockres->l_lock, flags);
3477 mlog(0, "lock %s\n", lockres->l_name);
3479 ret = ocfs2_dlm_unlock(osb->cconn, &lockres->l_lksb, lkm_flags);
3481 ocfs2_log_dlm_error("ocfs2_dlm_unlock", ret, lockres);
3482 mlog(ML_ERROR, "lockres flags: %lu\n", lockres->l_flags);
3483 ocfs2_dlm_dump_lksb(&lockres->l_lksb);
3486 mlog(0, "lock %s, successful return from ocfs2_dlm_unlock\n",
3489 ocfs2_wait_on_busy_lock(lockres);
3494 static void ocfs2_process_blocked_lock(struct ocfs2_super *osb,
3495 struct ocfs2_lock_res *lockres);
3497 /* Mark the lockres as being dropped. It will no longer be
3498 * queued if blocking, but we still may have to wait on it
3499 * being dequeued from the downconvert thread before we can consider
3502 * You can *not* attempt to call cluster_lock on this lockres anymore. */
3503 void ocfs2_mark_lockres_freeing(struct ocfs2_super *osb,
3504 struct ocfs2_lock_res *lockres)
3507 struct ocfs2_mask_waiter mw;
3508 unsigned long flags, flags2;
3510 ocfs2_init_mask_waiter(&mw);
3512 spin_lock_irqsave(&lockres->l_lock, flags);
3513 lockres->l_flags |= OCFS2_LOCK_FREEING;
3514 if (lockres->l_flags & OCFS2_LOCK_QUEUED && current == osb->dc_task) {
3516 * We know the downconvert is queued but not in progress
3517 * because we are the downconvert thread and processing
3518 * different lock. So we can just remove the lock from the
3519 * queue. This is not only an optimization but also a way
3520 * to avoid the following deadlock:
3521 * ocfs2_dentry_post_unlock()
3522 * ocfs2_dentry_lock_put()
3523 * ocfs2_drop_dentry_lock()
3525 * ocfs2_evict_inode()
3526 * ocfs2_clear_inode()
3527 * ocfs2_mark_lockres_freeing()
3528 * ... blocks waiting for OCFS2_LOCK_QUEUED
3529 * since we are the downconvert thread which
3530 * should clear the flag.
3532 spin_unlock_irqrestore(&lockres->l_lock, flags);
3533 spin_lock_irqsave(&osb->dc_task_lock, flags2);
3534 list_del_init(&lockres->l_blocked_list);
3535 osb->blocked_lock_count--;
3536 spin_unlock_irqrestore(&osb->dc_task_lock, flags2);
3538 * Warn if we recurse into another post_unlock call. Strictly
3539 * speaking it isn't a problem but we need to be careful if
3540 * that happens (stack overflow, deadlocks, ...) so warn if
3541 * ocfs2 grows a path for which this can happen.
3543 WARN_ON_ONCE(lockres->l_ops->post_unlock);
3544 /* Since the lock is freeing we don't do much in the fn below */
3545 ocfs2_process_blocked_lock(osb, lockres);
3548 while (lockres->l_flags & OCFS2_LOCK_QUEUED) {
3549 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_QUEUED, 0);
3550 spin_unlock_irqrestore(&lockres->l_lock, flags);
3552 mlog(0, "Waiting on lockres %s\n", lockres->l_name);
3554 status = ocfs2_wait_for_mask(&mw);
3558 spin_lock_irqsave(&lockres->l_lock, flags);
3560 spin_unlock_irqrestore(&lockres->l_lock, flags);
3563 void ocfs2_simple_drop_lockres(struct ocfs2_super *osb,
3564 struct ocfs2_lock_res *lockres)
3568 ocfs2_mark_lockres_freeing(osb, lockres);
3569 ret = ocfs2_drop_lock(osb, lockres);
3574 static void ocfs2_drop_osb_locks(struct ocfs2_super *osb)
3576 ocfs2_simple_drop_lockres(osb, &osb->osb_super_lockres);
3577 ocfs2_simple_drop_lockres(osb, &osb->osb_rename_lockres);
3578 ocfs2_simple_drop_lockres(osb, &osb->osb_nfs_sync_lockres);
3579 ocfs2_simple_drop_lockres(osb, &osb->osb_orphan_scan.os_lockres);
3582 int ocfs2_drop_inode_locks(struct inode *inode)
3586 /* No need to call ocfs2_mark_lockres_freeing here -
3587 * ocfs2_clear_inode has done it for us. */
3589 err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3590 &OCFS2_I(inode)->ip_open_lockres);
3596 err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3597 &OCFS2_I(inode)->ip_inode_lockres);
3600 if (err < 0 && !status)
3603 err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3604 &OCFS2_I(inode)->ip_rw_lockres);
3607 if (err < 0 && !status)
3613 static unsigned int ocfs2_prepare_downconvert(struct ocfs2_lock_res *lockres,
3616 assert_spin_locked(&lockres->l_lock);
3618 BUG_ON(lockres->l_blocking <= DLM_LOCK_NL);
3620 if (lockres->l_level <= new_level) {
3621 mlog(ML_ERROR, "lockres %s, lvl %d <= %d, blcklst %d, mask %d, "
3622 "type %d, flags 0x%lx, hold %d %d, act %d %d, req %d, "
3623 "block %d, pgen %d\n", lockres->l_name, lockres->l_level,
3624 new_level, list_empty(&lockres->l_blocked_list),
3625 list_empty(&lockres->l_mask_waiters), lockres->l_type,
3626 lockres->l_flags, lockres->l_ro_holders,
3627 lockres->l_ex_holders, lockres->l_action,
3628 lockres->l_unlock_action, lockres->l_requested,
3629 lockres->l_blocking, lockres->l_pending_gen);
3633 mlog(ML_BASTS, "lockres %s, level %d => %d, blocking %d\n",
3634 lockres->l_name, lockres->l_level, new_level, lockres->l_blocking);
3636 lockres->l_action = OCFS2_AST_DOWNCONVERT;
3637 lockres->l_requested = new_level;
3638 lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
3639 return lockres_set_pending(lockres);
3642 static int ocfs2_downconvert_lock(struct ocfs2_super *osb,
3643 struct ocfs2_lock_res *lockres,
3646 unsigned int generation)
3649 u32 dlm_flags = DLM_LKF_CONVERT;
3651 mlog(ML_BASTS, "lockres %s, level %d => %d\n", lockres->l_name,
3652 lockres->l_level, new_level);
3655 * On DLM_LKF_VALBLK, fsdlm behaves differently with o2cb. It always
3656 * expects DLM_LKF_VALBLK being set if the LKB has LVB, so that
3657 * we can recover correctly from node failure. Otherwise, we may get
3658 * invalid LVB in LKB, but without DLM_SBF_VALNOTVALID being set.
3660 if (ocfs2_userspace_stack(osb) &&
3661 lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
3665 dlm_flags |= DLM_LKF_VALBLK;
3667 ret = ocfs2_dlm_lock(osb->cconn,
3672 OCFS2_LOCK_ID_MAX_LEN - 1);
3673 lockres_clear_pending(lockres, generation, osb);
3675 ocfs2_log_dlm_error("ocfs2_dlm_lock", ret, lockres);
3676 ocfs2_recover_from_dlm_error(lockres, 1);
3685 /* returns 1 when the caller should unlock and call ocfs2_dlm_unlock */
3686 static int ocfs2_prepare_cancel_convert(struct ocfs2_super *osb,
3687 struct ocfs2_lock_res *lockres)
3689 assert_spin_locked(&lockres->l_lock);
3691 if (lockres->l_unlock_action == OCFS2_UNLOCK_CANCEL_CONVERT) {
3692 /* If we're already trying to cancel a lock conversion
3693 * then just drop the spinlock and allow the caller to
3694 * requeue this lock. */
3695 mlog(ML_BASTS, "lockres %s, skip convert\n", lockres->l_name);
3699 /* were we in a convert when we got the bast fire? */
3700 BUG_ON(lockres->l_action != OCFS2_AST_CONVERT &&
3701 lockres->l_action != OCFS2_AST_DOWNCONVERT);
3702 /* set things up for the unlockast to know to just
3703 * clear out the ast_action and unset busy, etc. */
3704 lockres->l_unlock_action = OCFS2_UNLOCK_CANCEL_CONVERT;
3706 mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_BUSY),
3707 "lock %s, invalid flags: 0x%lx\n",
3708 lockres->l_name, lockres->l_flags);
3710 mlog(ML_BASTS, "lockres %s\n", lockres->l_name);
3715 static int ocfs2_cancel_convert(struct ocfs2_super *osb,
3716 struct ocfs2_lock_res *lockres)
3720 ret = ocfs2_dlm_unlock(osb->cconn, &lockres->l_lksb,
3723 ocfs2_log_dlm_error("ocfs2_dlm_unlock", ret, lockres);
3724 ocfs2_recover_from_dlm_error(lockres, 0);
3727 mlog(ML_BASTS, "lockres %s\n", lockres->l_name);
3732 static int ocfs2_unblock_lock(struct ocfs2_super *osb,
3733 struct ocfs2_lock_res *lockres,
3734 struct ocfs2_unblock_ctl *ctl)
3736 unsigned long flags;
3744 spin_lock_irqsave(&lockres->l_lock, flags);
3748 * Is it still blocking? If not, we have no more work to do.
3750 if (!(lockres->l_flags & OCFS2_LOCK_BLOCKED)) {
3751 BUG_ON(lockres->l_blocking != DLM_LOCK_NL);
3752 spin_unlock_irqrestore(&lockres->l_lock, flags);
3757 if (lockres->l_flags & OCFS2_LOCK_BUSY) {
3759 * This is a *big* race. The OCFS2_LOCK_PENDING flag
3760 * exists entirely for one reason - another thread has set
3761 * OCFS2_LOCK_BUSY, but has *NOT* yet called dlm_lock().
3763 * If we do ocfs2_cancel_convert() before the other thread
3764 * calls dlm_lock(), our cancel will do nothing. We will
3765 * get no ast, and we will have no way of knowing the
3766 * cancel failed. Meanwhile, the other thread will call
3767 * into dlm_lock() and wait...forever.
3769 * Why forever? Because another node has asked for the
3770 * lock first; that's why we're here in unblock_lock().
3772 * The solution is OCFS2_LOCK_PENDING. When PENDING is
3773 * set, we just requeue the unblock. Only when the other
3774 * thread has called dlm_lock() and cleared PENDING will
3775 * we then cancel their request.
3777 * All callers of dlm_lock() must set OCFS2_DLM_PENDING
3778 * at the same time they set OCFS2_DLM_BUSY. They must
3779 * clear OCFS2_DLM_PENDING after dlm_lock() returns.
3781 if (lockres->l_flags & OCFS2_LOCK_PENDING) {
3782 mlog(ML_BASTS, "lockres %s, ReQ: Pending\n",
3788 ret = ocfs2_prepare_cancel_convert(osb, lockres);
3789 spin_unlock_irqrestore(&lockres->l_lock, flags);
3791 ret = ocfs2_cancel_convert(osb, lockres);
3799 * This prevents livelocks. OCFS2_LOCK_UPCONVERT_FINISHING flag is
3800 * set when the ast is received for an upconvert just before the
3801 * OCFS2_LOCK_BUSY flag is cleared. Now if the fs received a bast
3802 * on the heels of the ast, we want to delay the downconvert just
3803 * enough to allow the up requestor to do its task. Because this
3804 * lock is in the blocked queue, the lock will be downconverted
3805 * as soon as the requestor is done with the lock.
3807 if (lockres->l_flags & OCFS2_LOCK_UPCONVERT_FINISHING)
3811 * How can we block and yet be at NL? We were trying to upconvert
3812 * from NL and got canceled. The code comes back here, and now
3813 * we notice and clear BLOCKING.
3815 if (lockres->l_level == DLM_LOCK_NL) {
3816 BUG_ON(lockres->l_ex_holders || lockres->l_ro_holders);
3817 mlog(ML_BASTS, "lockres %s, Aborting dc\n", lockres->l_name);
3818 lockres->l_blocking = DLM_LOCK_NL;
3819 lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED);
3820 spin_unlock_irqrestore(&lockres->l_lock, flags);
3824 /* if we're blocking an exclusive and we have *any* holders,
3826 if ((lockres->l_blocking == DLM_LOCK_EX)
3827 && (lockres->l_ex_holders || lockres->l_ro_holders)) {
3828 mlog(ML_BASTS, "lockres %s, ReQ: EX/PR Holders %u,%u\n",
3829 lockres->l_name, lockres->l_ex_holders,
3830 lockres->l_ro_holders);
3834 /* If it's a PR we're blocking, then only
3835 * requeue if we've got any EX holders */
3836 if (lockres->l_blocking == DLM_LOCK_PR &&
3837 lockres->l_ex_holders) {
3838 mlog(ML_BASTS, "lockres %s, ReQ: EX Holders %u\n",
3839 lockres->l_name, lockres->l_ex_holders);
3844 * Can we get a lock in this state if the holder counts are
3845 * zero? The meta data unblock code used to check this.
3847 if ((lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
3848 && (lockres->l_flags & OCFS2_LOCK_REFRESHING)) {
3849 mlog(ML_BASTS, "lockres %s, ReQ: Lock Refreshing\n",
3854 new_level = ocfs2_highest_compat_lock_level(lockres->l_blocking);
3856 if (lockres->l_ops->check_downconvert
3857 && !lockres->l_ops->check_downconvert(lockres, new_level)) {
3858 mlog(ML_BASTS, "lockres %s, ReQ: Checkpointing\n",
3863 /* If we get here, then we know that there are no more
3864 * incompatible holders (and anyone asking for an incompatible
3865 * lock is blocked). We can now downconvert the lock */
3866 if (!lockres->l_ops->downconvert_worker)
3869 /* Some lockres types want to do a bit of work before
3870 * downconverting a lock. Allow that here. The worker function
3871 * may sleep, so we save off a copy of what we're blocking as
3872 * it may change while we're not holding the spin lock. */
3873 blocking = lockres->l_blocking;
3874 level = lockres->l_level;
3875 spin_unlock_irqrestore(&lockres->l_lock, flags);
3877 ctl->unblock_action = lockres->l_ops->downconvert_worker(lockres, blocking);
3879 if (ctl->unblock_action == UNBLOCK_STOP_POST) {
3880 mlog(ML_BASTS, "lockres %s, UNBLOCK_STOP_POST\n",
3885 spin_lock_irqsave(&lockres->l_lock, flags);
3886 if ((blocking != lockres->l_blocking) || (level != lockres->l_level)) {
3887 /* If this changed underneath us, then we can't drop
3889 mlog(ML_BASTS, "lockres %s, block=%d:%d, level=%d:%d, "
3890 "Recheck\n", lockres->l_name, blocking,
3891 lockres->l_blocking, level, lockres->l_level);
3898 if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
3899 if (lockres->l_level == DLM_LOCK_EX)
3903 * We only set the lvb if the lock has been fully
3904 * refreshed - otherwise we risk setting stale
3905 * data. Otherwise, there's no need to actually clear
3906 * out the lvb here as it's value is still valid.
3908 if (set_lvb && !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
3909 lockres->l_ops->set_lvb(lockres);
3912 gen = ocfs2_prepare_downconvert(lockres, new_level);
3913 spin_unlock_irqrestore(&lockres->l_lock, flags);
3914 ret = ocfs2_downconvert_lock(osb, lockres, new_level, set_lvb,
3916 /* The dlm lock convert is being cancelled in background,
3917 * ocfs2_cancel_convert() is asynchronous in fs/dlm,
3918 * requeue it, try again later.
3920 if (ret == -EBUSY) {
3922 mlog(ML_BASTS, "lockres %s, ReQ: Downconvert busy\n",
3934 spin_unlock_irqrestore(&lockres->l_lock, flags);
3940 static int ocfs2_data_convert_worker(struct ocfs2_lock_res *lockres,
3943 struct inode *inode;
3944 struct address_space *mapping;
3945 struct ocfs2_inode_info *oi;
3947 inode = ocfs2_lock_res_inode(lockres);
3948 mapping = inode->i_mapping;
3950 if (S_ISDIR(inode->i_mode)) {
3951 oi = OCFS2_I(inode);
3952 oi->ip_dir_lock_gen++;
3953 mlog(0, "generation: %u\n", oi->ip_dir_lock_gen);
3957 if (!S_ISREG(inode->i_mode))
3961 * We need this before the filemap_fdatawrite() so that it can
3962 * transfer the dirty bit from the PTE to the
3963 * page. Unfortunately this means that even for EX->PR
3964 * downconverts, we'll lose our mappings and have to build
3967 unmap_mapping_range(mapping, 0, 0, 0);
3969 if (filemap_fdatawrite(mapping)) {
3970 mlog(ML_ERROR, "Could not sync inode %llu for downconvert!",
3971 (unsigned long long)OCFS2_I(inode)->ip_blkno);
3973 sync_mapping_buffers(mapping);
3974 if (blocking == DLM_LOCK_EX) {
3975 truncate_inode_pages(mapping, 0);
3977 /* We only need to wait on the I/O if we're not also
3978 * truncating pages because truncate_inode_pages waits
3979 * for us above. We don't truncate pages if we're
3980 * blocking anything < EXMODE because we want to keep
3981 * them around in that case. */
3982 filemap_fdatawait(mapping);
3986 forget_all_cached_acls(inode);
3989 return UNBLOCK_CONTINUE;
3992 static int ocfs2_ci_checkpointed(struct ocfs2_caching_info *ci,
3993 struct ocfs2_lock_res *lockres,
3996 int checkpointed = ocfs2_ci_fully_checkpointed(ci);
3998 BUG_ON(new_level != DLM_LOCK_NL && new_level != DLM_LOCK_PR);
3999 BUG_ON(lockres->l_level != DLM_LOCK_EX && !checkpointed);
4004 ocfs2_start_checkpoint(OCFS2_SB(ocfs2_metadata_cache_get_super(ci)));
4008 static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
4011 struct inode *inode = ocfs2_lock_res_inode(lockres);
4013 return ocfs2_ci_checkpointed(INODE_CACHE(inode), lockres, new_level);
4016 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres)
4018 struct inode *inode = ocfs2_lock_res_inode(lockres);
4020 __ocfs2_stuff_meta_lvb(inode);
4024 * Does the final reference drop on our dentry lock. Right now this
4025 * happens in the downconvert thread, but we could choose to simplify the
4026 * dlmglue API and push these off to the ocfs2_wq in the future.
4028 static void ocfs2_dentry_post_unlock(struct ocfs2_super *osb,
4029 struct ocfs2_lock_res *lockres)
4031 struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
4032 ocfs2_dentry_lock_put(osb, dl);
4036 * d_delete() matching dentries before the lock downconvert.
4038 * At this point, any process waiting to destroy the
4039 * dentry_lock due to last ref count is stopped by the
4040 * OCFS2_LOCK_QUEUED flag.
4042 * We have two potential problems
4044 * 1) If we do the last reference drop on our dentry_lock (via dput)
4045 * we'll wind up in ocfs2_release_dentry_lock(), waiting on
4046 * the downconvert to finish. Instead we take an elevated
4047 * reference and push the drop until after we've completed our
4048 * unblock processing.
4050 * 2) There might be another process with a final reference,
4051 * waiting on us to finish processing. If this is the case, we
4052 * detect it and exit out - there's no more dentries anyway.
4054 static int ocfs2_dentry_convert_worker(struct ocfs2_lock_res *lockres,
4057 struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
4058 struct ocfs2_inode_info *oi = OCFS2_I(dl->dl_inode);
4059 struct dentry *dentry;
4060 unsigned long flags;
4064 * This node is blocking another node from getting a read
4065 * lock. This happens when we've renamed within a
4066 * directory. We've forced the other nodes to d_delete(), but
4067 * we never actually dropped our lock because it's still
4068 * valid. The downconvert code will retain a PR for this node,
4069 * so there's no further work to do.
4071 if (blocking == DLM_LOCK_PR)
4072 return UNBLOCK_CONTINUE;
4075 * Mark this inode as potentially orphaned. The code in
4076 * ocfs2_delete_inode() will figure out whether it actually
4077 * needs to be freed or not.
4079 spin_lock(&oi->ip_lock);
4080 oi->ip_flags |= OCFS2_INODE_MAYBE_ORPHANED;
4081 spin_unlock(&oi->ip_lock);
4084 * Yuck. We need to make sure however that the check of
4085 * OCFS2_LOCK_FREEING and the extra reference are atomic with
4086 * respect to a reference decrement or the setting of that
4089 spin_lock_irqsave(&lockres->l_lock, flags);
4090 spin_lock(&dentry_attach_lock);
4091 if (!(lockres->l_flags & OCFS2_LOCK_FREEING)
4096 spin_unlock(&dentry_attach_lock);
4097 spin_unlock_irqrestore(&lockres->l_lock, flags);
4099 mlog(0, "extra_ref = %d\n", extra_ref);
4102 * We have a process waiting on us in ocfs2_dentry_iput(),
4103 * which means we can't have any more outstanding
4104 * aliases. There's no need to do any more work.
4107 return UNBLOCK_CONTINUE;
4109 spin_lock(&dentry_attach_lock);
4111 dentry = ocfs2_find_local_alias(dl->dl_inode,
4112 dl->dl_parent_blkno, 1);
4115 spin_unlock(&dentry_attach_lock);
4117 if (S_ISDIR(dl->dl_inode->i_mode))
4118 shrink_dcache_parent(dentry);
4120 mlog(0, "d_delete(%pd);\n", dentry);
4123 * The following dcache calls may do an
4124 * iput(). Normally we don't want that from the
4125 * downconverting thread, but in this case it's ok
4126 * because the requesting node already has an
4127 * exclusive lock on the inode, so it can't be queued
4128 * for a downconvert.
4133 spin_lock(&dentry_attach_lock);
4135 spin_unlock(&dentry_attach_lock);
4138 * If we are the last holder of this dentry lock, there is no
4139 * reason to downconvert so skip straight to the unlock.
4141 if (dl->dl_count == 1)
4142 return UNBLOCK_STOP_POST;
4144 return UNBLOCK_CONTINUE_POST;
4147 static int ocfs2_check_refcount_downconvert(struct ocfs2_lock_res *lockres,
4150 struct ocfs2_refcount_tree *tree =
4151 ocfs2_lock_res_refcount_tree(lockres);
4153 return ocfs2_ci_checkpointed(&tree->rf_ci, lockres, new_level);
4156 static int ocfs2_refcount_convert_worker(struct ocfs2_lock_res *lockres,
4159 struct ocfs2_refcount_tree *tree =
4160 ocfs2_lock_res_refcount_tree(lockres);
4162 ocfs2_metadata_cache_purge(&tree->rf_ci);
4164 return UNBLOCK_CONTINUE;
4167 static void ocfs2_set_qinfo_lvb(struct ocfs2_lock_res *lockres)
4169 struct ocfs2_qinfo_lvb *lvb;
4170 struct ocfs2_mem_dqinfo *oinfo = ocfs2_lock_res_qinfo(lockres);
4171 struct mem_dqinfo *info = sb_dqinfo(oinfo->dqi_gi.dqi_sb,
4172 oinfo->dqi_gi.dqi_type);
4174 lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
4175 lvb->lvb_version = OCFS2_QINFO_LVB_VERSION;
4176 lvb->lvb_bgrace = cpu_to_be32(info->dqi_bgrace);
4177 lvb->lvb_igrace = cpu_to_be32(info->dqi_igrace);
4178 lvb->lvb_syncms = cpu_to_be32(oinfo->dqi_syncms);
4179 lvb->lvb_blocks = cpu_to_be32(oinfo->dqi_gi.dqi_blocks);
4180 lvb->lvb_free_blk = cpu_to_be32(oinfo->dqi_gi.dqi_free_blk);
4181 lvb->lvb_free_entry = cpu_to_be32(oinfo->dqi_gi.dqi_free_entry);
4184 void ocfs2_qinfo_unlock(struct ocfs2_mem_dqinfo *oinfo, int ex)
4186 struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
4187 struct ocfs2_super *osb = OCFS2_SB(oinfo->dqi_gi.dqi_sb);
4188 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4190 if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb))
4191 ocfs2_cluster_unlock(osb, lockres, level);
4194 static int ocfs2_refresh_qinfo(struct ocfs2_mem_dqinfo *oinfo)
4196 struct mem_dqinfo *info = sb_dqinfo(oinfo->dqi_gi.dqi_sb,
4197 oinfo->dqi_gi.dqi_type);
4198 struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
4199 struct ocfs2_qinfo_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
4200 struct buffer_head *bh = NULL;
4201 struct ocfs2_global_disk_dqinfo *gdinfo;
4204 if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
4205 lvb->lvb_version == OCFS2_QINFO_LVB_VERSION) {
4206 info->dqi_bgrace = be32_to_cpu(lvb->lvb_bgrace);
4207 info->dqi_igrace = be32_to_cpu(lvb->lvb_igrace);
4208 oinfo->dqi_syncms = be32_to_cpu(lvb->lvb_syncms);
4209 oinfo->dqi_gi.dqi_blocks = be32_to_cpu(lvb->lvb_blocks);
4210 oinfo->dqi_gi.dqi_free_blk = be32_to_cpu(lvb->lvb_free_blk);
4211 oinfo->dqi_gi.dqi_free_entry =
4212 be32_to_cpu(lvb->lvb_free_entry);
4214 status = ocfs2_read_quota_phys_block(oinfo->dqi_gqinode,
4215 oinfo->dqi_giblk, &bh);
4220 gdinfo = (struct ocfs2_global_disk_dqinfo *)
4221 (bh->b_data + OCFS2_GLOBAL_INFO_OFF);
4222 info->dqi_bgrace = le32_to_cpu(gdinfo->dqi_bgrace);
4223 info->dqi_igrace = le32_to_cpu(gdinfo->dqi_igrace);
4224 oinfo->dqi_syncms = le32_to_cpu(gdinfo->dqi_syncms);
4225 oinfo->dqi_gi.dqi_blocks = le32_to_cpu(gdinfo->dqi_blocks);
4226 oinfo->dqi_gi.dqi_free_blk = le32_to_cpu(gdinfo->dqi_free_blk);
4227 oinfo->dqi_gi.dqi_free_entry =
4228 le32_to_cpu(gdinfo->dqi_free_entry);
4230 ocfs2_track_lock_refresh(lockres);
4237 /* Lock quota info, this function expects at least shared lock on the quota file
4238 * so that we can safely refresh quota info from disk. */
4239 int ocfs2_qinfo_lock(struct ocfs2_mem_dqinfo *oinfo, int ex)
4241 struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
4242 struct ocfs2_super *osb = OCFS2_SB(oinfo->dqi_gi.dqi_sb);
4243 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4246 /* On RO devices, locking really isn't needed... */
4247 if (ocfs2_is_hard_readonly(osb)) {
4252 if (ocfs2_mount_local(osb))
4255 status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
4260 if (!ocfs2_should_refresh_lock_res(lockres))
4262 /* OK, we have the lock but we need to refresh the quota info */
4263 status = ocfs2_refresh_qinfo(oinfo);
4265 ocfs2_qinfo_unlock(oinfo, ex);
4266 ocfs2_complete_lock_res_refresh(lockres, status);
4271 int ocfs2_refcount_lock(struct ocfs2_refcount_tree *ref_tree, int ex)
4274 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4275 struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
4276 struct ocfs2_super *osb = lockres->l_priv;
4279 if (ocfs2_is_hard_readonly(osb))
4282 if (ocfs2_mount_local(osb))
4285 status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
4292 void ocfs2_refcount_unlock(struct ocfs2_refcount_tree *ref_tree, int ex)
4294 int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4295 struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
4296 struct ocfs2_super *osb = lockres->l_priv;
4298 if (!ocfs2_mount_local(osb))
4299 ocfs2_cluster_unlock(osb, lockres, level);
4302 static void ocfs2_process_blocked_lock(struct ocfs2_super *osb,
4303 struct ocfs2_lock_res *lockres)
4306 struct ocfs2_unblock_ctl ctl = {0, 0,};
4307 unsigned long flags;
4309 /* Our reference to the lockres in this function can be
4310 * considered valid until we remove the OCFS2_LOCK_QUEUED
4314 BUG_ON(!lockres->l_ops);
4316 mlog(ML_BASTS, "lockres %s blocked\n", lockres->l_name);
4318 /* Detect whether a lock has been marked as going away while
4319 * the downconvert thread was processing other things. A lock can
4320 * still be marked with OCFS2_LOCK_FREEING after this check,
4321 * but short circuiting here will still save us some
4323 spin_lock_irqsave(&lockres->l_lock, flags);
4324 if (lockres->l_flags & OCFS2_LOCK_FREEING)
4326 spin_unlock_irqrestore(&lockres->l_lock, flags);
4328 status = ocfs2_unblock_lock(osb, lockres, &ctl);
4332 spin_lock_irqsave(&lockres->l_lock, flags);
4334 if (lockres->l_flags & OCFS2_LOCK_FREEING || !ctl.requeue) {
4335 lockres_clear_flags(lockres, OCFS2_LOCK_QUEUED);
4337 ocfs2_schedule_blocked_lock(osb, lockres);
4339 mlog(ML_BASTS, "lockres %s, requeue = %s.\n", lockres->l_name,
4340 ctl.requeue ? "yes" : "no");
4341 spin_unlock_irqrestore(&lockres->l_lock, flags);
4343 if (ctl.unblock_action != UNBLOCK_CONTINUE
4344 && lockres->l_ops->post_unlock)
4345 lockres->l_ops->post_unlock(osb, lockres);
4348 static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
4349 struct ocfs2_lock_res *lockres)
4351 unsigned long flags;
4353 assert_spin_locked(&lockres->l_lock);
4355 if (lockres->l_flags & OCFS2_LOCK_FREEING) {
4356 /* Do not schedule a lock for downconvert when it's on
4357 * the way to destruction - any nodes wanting access
4358 * to the resource will get it soon. */
4359 mlog(ML_BASTS, "lockres %s won't be scheduled: flags 0x%lx\n",
4360 lockres->l_name, lockres->l_flags);
4364 lockres_or_flags(lockres, OCFS2_LOCK_QUEUED);
4366 spin_lock_irqsave(&osb->dc_task_lock, flags);
4367 if (list_empty(&lockres->l_blocked_list)) {
4368 list_add_tail(&lockres->l_blocked_list,
4369 &osb->blocked_lock_list);
4370 osb->blocked_lock_count++;
4372 spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4375 static void ocfs2_downconvert_thread_do_work(struct ocfs2_super *osb)
4377 unsigned long processed;
4378 unsigned long flags;
4379 struct ocfs2_lock_res *lockres;
4381 spin_lock_irqsave(&osb->dc_task_lock, flags);
4382 /* grab this early so we know to try again if a state change and
4383 * wake happens part-way through our work */
4384 osb->dc_work_sequence = osb->dc_wake_sequence;
4386 processed = osb->blocked_lock_count;
4388 * blocked lock processing in this loop might call iput which can
4389 * remove items off osb->blocked_lock_list. Downconvert up to
4390 * 'processed' number of locks, but stop short if we had some
4391 * removed in ocfs2_mark_lockres_freeing when downconverting.
4393 while (processed && !list_empty(&osb->blocked_lock_list)) {
4394 lockres = list_entry(osb->blocked_lock_list.next,
4395 struct ocfs2_lock_res, l_blocked_list);
4396 list_del_init(&lockres->l_blocked_list);
4397 osb->blocked_lock_count--;
4398 spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4403 ocfs2_process_blocked_lock(osb, lockres);
4405 spin_lock_irqsave(&osb->dc_task_lock, flags);
4407 spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4410 static int ocfs2_downconvert_thread_lists_empty(struct ocfs2_super *osb)
4413 unsigned long flags;
4415 spin_lock_irqsave(&osb->dc_task_lock, flags);
4416 if (list_empty(&osb->blocked_lock_list))
4419 spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4423 static int ocfs2_downconvert_thread_should_wake(struct ocfs2_super *osb)
4425 int should_wake = 0;
4426 unsigned long flags;
4428 spin_lock_irqsave(&osb->dc_task_lock, flags);
4429 if (osb->dc_work_sequence != osb->dc_wake_sequence)
4431 spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4436 static int ocfs2_downconvert_thread(void *arg)
4438 struct ocfs2_super *osb = arg;
4440 /* only quit once we've been asked to stop and there is no more
4442 while (!(kthread_should_stop() &&
4443 ocfs2_downconvert_thread_lists_empty(osb))) {
4445 wait_event_interruptible(osb->dc_event,
4446 ocfs2_downconvert_thread_should_wake(osb) ||
4447 kthread_should_stop());
4449 mlog(0, "downconvert_thread: awoken\n");
4451 ocfs2_downconvert_thread_do_work(osb);
4454 osb->dc_task = NULL;
4458 void ocfs2_wake_downconvert_thread(struct ocfs2_super *osb)
4460 unsigned long flags;
4462 spin_lock_irqsave(&osb->dc_task_lock, flags);
4463 /* make sure the voting thread gets a swipe at whatever changes
4464 * the caller may have made to the voting state */
4465 osb->dc_wake_sequence++;
4466 spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4467 wake_up(&osb->dc_event);