xref: /openbmc/linux/fs/ocfs2/dlmglue.c (revision aac5987a)
1 /* -*- mode: c; c-basic-offset: 8; -*-
2  * vim: noexpandtab sw=8 ts=8 sts=0:
3  *
4  * dlmglue.c
5  *
6  * Code which implements an OCFS2 specific interface to our DLM.
7  *
8  * Copyright (C) 2003, 2004 Oracle.  All rights reserved.
9  *
10  * This program is free software; you can redistribute it and/or
11  * modify it under the terms of the GNU General Public
12  * License as published by the Free Software Foundation; either
13  * version 2 of the License, or (at your option) any later version.
14  *
15  * This program is distributed in the hope that it will be useful,
16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
18  * General Public License for more details.
19  *
20  * You should have received a copy of the GNU General Public
21  * License along with this program; if not, write to the
22  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
23  * Boston, MA 021110-1307, USA.
24  */
25 
26 #include <linux/types.h>
27 #include <linux/slab.h>
28 #include <linux/highmem.h>
29 #include <linux/mm.h>
30 #include <linux/kthread.h>
31 #include <linux/pagemap.h>
32 #include <linux/debugfs.h>
33 #include <linux/seq_file.h>
34 #include <linux/time.h>
35 #include <linux/quotaops.h>
36 #include <linux/sched/signal.h>
37 
38 #define MLOG_MASK_PREFIX ML_DLM_GLUE
39 #include <cluster/masklog.h>
40 
41 #include "ocfs2.h"
42 #include "ocfs2_lockingver.h"
43 
44 #include "alloc.h"
45 #include "dcache.h"
46 #include "dlmglue.h"
47 #include "extent_map.h"
48 #include "file.h"
49 #include "heartbeat.h"
50 #include "inode.h"
51 #include "journal.h"
52 #include "stackglue.h"
53 #include "slot_map.h"
54 #include "super.h"
55 #include "uptodate.h"
56 #include "quota.h"
57 #include "refcounttree.h"
58 #include "acl.h"
59 
60 #include "buffer_head_io.h"
61 
62 struct ocfs2_mask_waiter {
63 	struct list_head	mw_item;
64 	int			mw_status;
65 	struct completion	mw_complete;
66 	unsigned long		mw_mask;
67 	unsigned long		mw_goal;
68 #ifdef CONFIG_OCFS2_FS_STATS
69 	ktime_t			mw_lock_start;
70 #endif
71 };
72 
73 static struct ocfs2_super *ocfs2_get_dentry_osb(struct ocfs2_lock_res *lockres);
74 static struct ocfs2_super *ocfs2_get_inode_osb(struct ocfs2_lock_res *lockres);
75 static struct ocfs2_super *ocfs2_get_file_osb(struct ocfs2_lock_res *lockres);
76 static struct ocfs2_super *ocfs2_get_qinfo_osb(struct ocfs2_lock_res *lockres);
77 
78 /*
79  * Return value from ->downconvert_worker functions.
80  *
81  * These control the precise actions of ocfs2_unblock_lock()
82  * and ocfs2_process_blocked_lock()
83  *
84  */
85 enum ocfs2_unblock_action {
86 	UNBLOCK_CONTINUE	= 0, /* Continue downconvert */
87 	UNBLOCK_CONTINUE_POST	= 1, /* Continue downconvert, fire
88 				      * ->post_unlock callback */
89 	UNBLOCK_STOP_POST	= 2, /* Do not downconvert, fire
90 				      * ->post_unlock() callback. */
91 };
92 
93 struct ocfs2_unblock_ctl {
94 	int requeue;
95 	enum ocfs2_unblock_action unblock_action;
96 };
97 
98 /* Lockdep class keys */
99 struct lock_class_key lockdep_keys[OCFS2_NUM_LOCK_TYPES];
100 
101 static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
102 					int new_level);
103 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres);
104 
105 static int ocfs2_data_convert_worker(struct ocfs2_lock_res *lockres,
106 				     int blocking);
107 
108 static int ocfs2_dentry_convert_worker(struct ocfs2_lock_res *lockres,
109 				       int blocking);
110 
111 static void ocfs2_dentry_post_unlock(struct ocfs2_super *osb,
112 				     struct ocfs2_lock_res *lockres);
113 
114 static void ocfs2_set_qinfo_lvb(struct ocfs2_lock_res *lockres);
115 
116 static int ocfs2_check_refcount_downconvert(struct ocfs2_lock_res *lockres,
117 					    int new_level);
118 static int ocfs2_refcount_convert_worker(struct ocfs2_lock_res *lockres,
119 					 int blocking);
120 
121 #define mlog_meta_lvb(__level, __lockres) ocfs2_dump_meta_lvb_info(__level, __PRETTY_FUNCTION__, __LINE__, __lockres)
122 
123 /* This aids in debugging situations where a bad LVB might be involved. */
124 static void ocfs2_dump_meta_lvb_info(u64 level,
125 				     const char *function,
126 				     unsigned int line,
127 				     struct ocfs2_lock_res *lockres)
128 {
129 	struct ocfs2_meta_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
130 
131 	mlog(level, "LVB information for %s (called from %s:%u):\n",
132 	     lockres->l_name, function, line);
133 	mlog(level, "version: %u, clusters: %u, generation: 0x%x\n",
134 	     lvb->lvb_version, be32_to_cpu(lvb->lvb_iclusters),
135 	     be32_to_cpu(lvb->lvb_igeneration));
136 	mlog(level, "size: %llu, uid %u, gid %u, mode 0x%x\n",
137 	     (unsigned long long)be64_to_cpu(lvb->lvb_isize),
138 	     be32_to_cpu(lvb->lvb_iuid), be32_to_cpu(lvb->lvb_igid),
139 	     be16_to_cpu(lvb->lvb_imode));
140 	mlog(level, "nlink %u, atime_packed 0x%llx, ctime_packed 0x%llx, "
141 	     "mtime_packed 0x%llx iattr 0x%x\n", be16_to_cpu(lvb->lvb_inlink),
142 	     (long long)be64_to_cpu(lvb->lvb_iatime_packed),
143 	     (long long)be64_to_cpu(lvb->lvb_ictime_packed),
144 	     (long long)be64_to_cpu(lvb->lvb_imtime_packed),
145 	     be32_to_cpu(lvb->lvb_iattr));
146 }
147 
148 
149 /*
150  * OCFS2 Lock Resource Operations
151  *
152  * These fine tune the behavior of the generic dlmglue locking infrastructure.
153  *
154  * The most basic of lock types can point ->l_priv to their respective
155  * struct ocfs2_super and allow the default actions to manage things.
156  *
157  * Right now, each lock type also needs to implement an init function,
158  * and trivial lock/unlock wrappers. ocfs2_simple_drop_lockres()
159  * should be called when the lock is no longer needed (i.e., object
160  * destruction time).
161  */
162 struct ocfs2_lock_res_ops {
163 	/*
164 	 * Translate an ocfs2_lock_res * into an ocfs2_super *. Define
165 	 * this callback if ->l_priv is not an ocfs2_super pointer
166 	 */
167 	struct ocfs2_super * (*get_osb)(struct ocfs2_lock_res *);
168 
169 	/*
170 	 * Optionally called in the downconvert thread after a
171 	 * successful downconvert. The lockres will not be referenced
172 	 * after this callback is called, so it is safe to free
173 	 * memory, etc.
174 	 *
175 	 * The exact semantics of when this is called are controlled
176 	 * by ->downconvert_worker()
177 	 */
178 	void (*post_unlock)(struct ocfs2_super *, struct ocfs2_lock_res *);
179 
180 	/*
181 	 * Allow a lock type to add checks to determine whether it is
182 	 * safe to downconvert a lock. Return 0 to re-queue the
183 	 * downconvert at a later time, nonzero to continue.
184 	 *
185 	 * For most locks, the default checks that there are no
186 	 * incompatible holders are sufficient.
187 	 *
188 	 * Called with the lockres spinlock held.
189 	 */
190 	int (*check_downconvert)(struct ocfs2_lock_res *, int);
191 
192 	/*
193 	 * Allows a lock type to populate the lock value block. This
194 	 * is called on downconvert, and when we drop a lock.
195 	 *
196 	 * Locks that want to use this should set LOCK_TYPE_USES_LVB
197 	 * in the flags field.
198 	 *
199 	 * Called with the lockres spinlock held.
200 	 */
201 	void (*set_lvb)(struct ocfs2_lock_res *);
202 
203 	/*
204 	 * Called from the downconvert thread when it is determined
205 	 * that a lock will be downconverted. This is called without
206 	 * any locks held so the function can do work that might
207 	 * schedule (syncing out data, etc).
208 	 *
209 	 * This should return any one of the ocfs2_unblock_action
210 	 * values, depending on what it wants the thread to do.
211 	 */
212 	int (*downconvert_worker)(struct ocfs2_lock_res *, int);
213 
214 	/*
215 	 * LOCK_TYPE_* flags which describe the specific requirements
216 	 * of a lock type. Descriptions of each individual flag follow.
217 	 */
218 	int flags;
219 };
220 
221 /*
222  * Some locks want to "refresh" potentially stale data when a
223  * meaningful (PRMODE or EXMODE) lock level is first obtained. If this
224  * flag is set, the OCFS2_LOCK_NEEDS_REFRESH flag will be set on the
225  * individual lockres l_flags member from the ast function. It is
226  * expected that the locking wrapper will clear the
227  * OCFS2_LOCK_NEEDS_REFRESH flag when done.
228  */
229 #define LOCK_TYPE_REQUIRES_REFRESH 0x1
230 
231 /*
232  * Indicate that a lock type makes use of the lock value block. The
233  * ->set_lvb lock type callback must be defined.
234  */
235 #define LOCK_TYPE_USES_LVB		0x2
236 
237 static struct ocfs2_lock_res_ops ocfs2_inode_rw_lops = {
238 	.get_osb	= ocfs2_get_inode_osb,
239 	.flags		= 0,
240 };
241 
242 static struct ocfs2_lock_res_ops ocfs2_inode_inode_lops = {
243 	.get_osb	= ocfs2_get_inode_osb,
244 	.check_downconvert = ocfs2_check_meta_downconvert,
245 	.set_lvb	= ocfs2_set_meta_lvb,
246 	.downconvert_worker = ocfs2_data_convert_worker,
247 	.flags		= LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_USES_LVB,
248 };
249 
250 static struct ocfs2_lock_res_ops ocfs2_super_lops = {
251 	.flags		= LOCK_TYPE_REQUIRES_REFRESH,
252 };
253 
254 static struct ocfs2_lock_res_ops ocfs2_rename_lops = {
255 	.flags		= 0,
256 };
257 
258 static struct ocfs2_lock_res_ops ocfs2_nfs_sync_lops = {
259 	.flags		= 0,
260 };
261 
262 static struct ocfs2_lock_res_ops ocfs2_orphan_scan_lops = {
263 	.flags		= LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_USES_LVB,
264 };
265 
266 static struct ocfs2_lock_res_ops ocfs2_dentry_lops = {
267 	.get_osb	= ocfs2_get_dentry_osb,
268 	.post_unlock	= ocfs2_dentry_post_unlock,
269 	.downconvert_worker = ocfs2_dentry_convert_worker,
270 	.flags		= 0,
271 };
272 
273 static struct ocfs2_lock_res_ops ocfs2_inode_open_lops = {
274 	.get_osb	= ocfs2_get_inode_osb,
275 	.flags		= 0,
276 };
277 
278 static struct ocfs2_lock_res_ops ocfs2_flock_lops = {
279 	.get_osb	= ocfs2_get_file_osb,
280 	.flags		= 0,
281 };
282 
283 static struct ocfs2_lock_res_ops ocfs2_qinfo_lops = {
284 	.set_lvb	= ocfs2_set_qinfo_lvb,
285 	.get_osb	= ocfs2_get_qinfo_osb,
286 	.flags		= LOCK_TYPE_REQUIRES_REFRESH | LOCK_TYPE_USES_LVB,
287 };
288 
289 static struct ocfs2_lock_res_ops ocfs2_refcount_block_lops = {
290 	.check_downconvert = ocfs2_check_refcount_downconvert,
291 	.downconvert_worker = ocfs2_refcount_convert_worker,
292 	.flags		= 0,
293 };
294 
295 static inline int ocfs2_is_inode_lock(struct ocfs2_lock_res *lockres)
296 {
297 	return lockres->l_type == OCFS2_LOCK_TYPE_META ||
298 		lockres->l_type == OCFS2_LOCK_TYPE_RW ||
299 		lockres->l_type == OCFS2_LOCK_TYPE_OPEN;
300 }
301 
302 static inline struct ocfs2_lock_res *ocfs2_lksb_to_lock_res(struct ocfs2_dlm_lksb *lksb)
303 {
304 	return container_of(lksb, struct ocfs2_lock_res, l_lksb);
305 }
306 
307 static inline struct inode *ocfs2_lock_res_inode(struct ocfs2_lock_res *lockres)
308 {
309 	BUG_ON(!ocfs2_is_inode_lock(lockres));
310 
311 	return (struct inode *) lockres->l_priv;
312 }
313 
314 static inline struct ocfs2_dentry_lock *ocfs2_lock_res_dl(struct ocfs2_lock_res *lockres)
315 {
316 	BUG_ON(lockres->l_type != OCFS2_LOCK_TYPE_DENTRY);
317 
318 	return (struct ocfs2_dentry_lock *)lockres->l_priv;
319 }
320 
321 static inline struct ocfs2_mem_dqinfo *ocfs2_lock_res_qinfo(struct ocfs2_lock_res *lockres)
322 {
323 	BUG_ON(lockres->l_type != OCFS2_LOCK_TYPE_QINFO);
324 
325 	return (struct ocfs2_mem_dqinfo *)lockres->l_priv;
326 }
327 
328 static inline struct ocfs2_refcount_tree *
329 ocfs2_lock_res_refcount_tree(struct ocfs2_lock_res *res)
330 {
331 	return container_of(res, struct ocfs2_refcount_tree, rf_lockres);
332 }
333 
334 static inline struct ocfs2_super *ocfs2_get_lockres_osb(struct ocfs2_lock_res *lockres)
335 {
336 	if (lockres->l_ops->get_osb)
337 		return lockres->l_ops->get_osb(lockres);
338 
339 	return (struct ocfs2_super *)lockres->l_priv;
340 }
341 
342 static int ocfs2_lock_create(struct ocfs2_super *osb,
343 			     struct ocfs2_lock_res *lockres,
344 			     int level,
345 			     u32 dlm_flags);
346 static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lockres,
347 						     int wanted);
348 static void __ocfs2_cluster_unlock(struct ocfs2_super *osb,
349 				   struct ocfs2_lock_res *lockres,
350 				   int level, unsigned long caller_ip);
351 static inline void ocfs2_cluster_unlock(struct ocfs2_super *osb,
352 					struct ocfs2_lock_res *lockres,
353 					int level)
354 {
355 	__ocfs2_cluster_unlock(osb, lockres, level, _RET_IP_);
356 }
357 
358 static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res *lockres);
359 static inline void ocfs2_generic_handle_convert_action(struct ocfs2_lock_res *lockres);
360 static inline void ocfs2_generic_handle_attach_action(struct ocfs2_lock_res *lockres);
361 static int ocfs2_generic_handle_bast(struct ocfs2_lock_res *lockres, int level);
362 static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
363 					struct ocfs2_lock_res *lockres);
364 static inline void ocfs2_recover_from_dlm_error(struct ocfs2_lock_res *lockres,
365 						int convert);
366 #define ocfs2_log_dlm_error(_func, _err, _lockres) do {					\
367 	if ((_lockres)->l_type != OCFS2_LOCK_TYPE_DENTRY)				\
368 		mlog(ML_ERROR, "DLM error %d while calling %s on resource %s\n",	\
369 		     _err, _func, _lockres->l_name);					\
370 	else										\
371 		mlog(ML_ERROR, "DLM error %d while calling %s on resource %.*s%08x\n",	\
372 		     _err, _func, OCFS2_DENTRY_LOCK_INO_START - 1, (_lockres)->l_name,	\
373 		     (unsigned int)ocfs2_get_dentry_lock_ino(_lockres));		\
374 } while (0)
375 static int ocfs2_downconvert_thread(void *arg);
376 static void ocfs2_downconvert_on_unlock(struct ocfs2_super *osb,
377 					struct ocfs2_lock_res *lockres);
378 static int ocfs2_inode_lock_update(struct inode *inode,
379 				  struct buffer_head **bh);
380 static void ocfs2_drop_osb_locks(struct ocfs2_super *osb);
381 static inline int ocfs2_highest_compat_lock_level(int level);
382 static unsigned int ocfs2_prepare_downconvert(struct ocfs2_lock_res *lockres,
383 					      int new_level);
384 static int ocfs2_downconvert_lock(struct ocfs2_super *osb,
385 				  struct ocfs2_lock_res *lockres,
386 				  int new_level,
387 				  int lvb,
388 				  unsigned int generation);
389 static int ocfs2_prepare_cancel_convert(struct ocfs2_super *osb,
390 				        struct ocfs2_lock_res *lockres);
391 static int ocfs2_cancel_convert(struct ocfs2_super *osb,
392 				struct ocfs2_lock_res *lockres);
393 
394 
395 static void ocfs2_build_lock_name(enum ocfs2_lock_type type,
396 				  u64 blkno,
397 				  u32 generation,
398 				  char *name)
399 {
400 	int len;
401 
402 	BUG_ON(type >= OCFS2_NUM_LOCK_TYPES);
403 
404 	len = snprintf(name, OCFS2_LOCK_ID_MAX_LEN, "%c%s%016llx%08x",
405 		       ocfs2_lock_type_char(type), OCFS2_LOCK_ID_PAD,
406 		       (long long)blkno, generation);
407 
408 	BUG_ON(len != (OCFS2_LOCK_ID_MAX_LEN - 1));
409 
410 	mlog(0, "built lock resource with name: %s\n", name);
411 }
412 
413 static DEFINE_SPINLOCK(ocfs2_dlm_tracking_lock);
414 
415 static void ocfs2_add_lockres_tracking(struct ocfs2_lock_res *res,
416 				       struct ocfs2_dlm_debug *dlm_debug)
417 {
418 	mlog(0, "Add tracking for lockres %s\n", res->l_name);
419 
420 	spin_lock(&ocfs2_dlm_tracking_lock);
421 	list_add(&res->l_debug_list, &dlm_debug->d_lockres_tracking);
422 	spin_unlock(&ocfs2_dlm_tracking_lock);
423 }
424 
425 static void ocfs2_remove_lockres_tracking(struct ocfs2_lock_res *res)
426 {
427 	spin_lock(&ocfs2_dlm_tracking_lock);
428 	if (!list_empty(&res->l_debug_list))
429 		list_del_init(&res->l_debug_list);
430 	spin_unlock(&ocfs2_dlm_tracking_lock);
431 }
432 
433 #ifdef CONFIG_OCFS2_FS_STATS
434 static void ocfs2_init_lock_stats(struct ocfs2_lock_res *res)
435 {
436 	res->l_lock_refresh = 0;
437 	memset(&res->l_lock_prmode, 0, sizeof(struct ocfs2_lock_stats));
438 	memset(&res->l_lock_exmode, 0, sizeof(struct ocfs2_lock_stats));
439 }
440 
441 static void ocfs2_update_lock_stats(struct ocfs2_lock_res *res, int level,
442 				    struct ocfs2_mask_waiter *mw, int ret)
443 {
444 	u32 usec;
445 	ktime_t kt;
446 	struct ocfs2_lock_stats *stats;
447 
448 	if (level == LKM_PRMODE)
449 		stats = &res->l_lock_prmode;
450 	else if (level == LKM_EXMODE)
451 		stats = &res->l_lock_exmode;
452 	else
453 		return;
454 
455 	kt = ktime_sub(ktime_get(), mw->mw_lock_start);
456 	usec = ktime_to_us(kt);
457 
458 	stats->ls_gets++;
459 	stats->ls_total += ktime_to_ns(kt);
460 	/* overflow */
461 	if (unlikely(stats->ls_gets == 0)) {
462 		stats->ls_gets++;
463 		stats->ls_total = ktime_to_ns(kt);
464 	}
465 
466 	if (stats->ls_max < usec)
467 		stats->ls_max = usec;
468 
469 	if (ret)
470 		stats->ls_fail++;
471 }
472 
473 static inline void ocfs2_track_lock_refresh(struct ocfs2_lock_res *lockres)
474 {
475 	lockres->l_lock_refresh++;
476 }
477 
478 static inline void ocfs2_init_start_time(struct ocfs2_mask_waiter *mw)
479 {
480 	mw->mw_lock_start = ktime_get();
481 }
482 #else
483 static inline void ocfs2_init_lock_stats(struct ocfs2_lock_res *res)
484 {
485 }
486 static inline void ocfs2_update_lock_stats(struct ocfs2_lock_res *res,
487 			   int level, struct ocfs2_mask_waiter *mw, int ret)
488 {
489 }
490 static inline void ocfs2_track_lock_refresh(struct ocfs2_lock_res *lockres)
491 {
492 }
493 static inline void ocfs2_init_start_time(struct ocfs2_mask_waiter *mw)
494 {
495 }
496 #endif
497 
498 static void ocfs2_lock_res_init_common(struct ocfs2_super *osb,
499 				       struct ocfs2_lock_res *res,
500 				       enum ocfs2_lock_type type,
501 				       struct ocfs2_lock_res_ops *ops,
502 				       void *priv)
503 {
504 	res->l_type          = type;
505 	res->l_ops           = ops;
506 	res->l_priv          = priv;
507 
508 	res->l_level         = DLM_LOCK_IV;
509 	res->l_requested     = DLM_LOCK_IV;
510 	res->l_blocking      = DLM_LOCK_IV;
511 	res->l_action        = OCFS2_AST_INVALID;
512 	res->l_unlock_action = OCFS2_UNLOCK_INVALID;
513 
514 	res->l_flags         = OCFS2_LOCK_INITIALIZED;
515 
516 	ocfs2_add_lockres_tracking(res, osb->osb_dlm_debug);
517 
518 	ocfs2_init_lock_stats(res);
519 #ifdef CONFIG_DEBUG_LOCK_ALLOC
520 	if (type != OCFS2_LOCK_TYPE_OPEN)
521 		lockdep_init_map(&res->l_lockdep_map, ocfs2_lock_type_strings[type],
522 				 &lockdep_keys[type], 0);
523 	else
524 		res->l_lockdep_map.key = NULL;
525 #endif
526 }
527 
528 void ocfs2_lock_res_init_once(struct ocfs2_lock_res *res)
529 {
530 	/* This also clears out the lock status block */
531 	memset(res, 0, sizeof(struct ocfs2_lock_res));
532 	spin_lock_init(&res->l_lock);
533 	init_waitqueue_head(&res->l_event);
534 	INIT_LIST_HEAD(&res->l_blocked_list);
535 	INIT_LIST_HEAD(&res->l_mask_waiters);
536 	INIT_LIST_HEAD(&res->l_holders);
537 }
538 
539 void ocfs2_inode_lock_res_init(struct ocfs2_lock_res *res,
540 			       enum ocfs2_lock_type type,
541 			       unsigned int generation,
542 			       struct inode *inode)
543 {
544 	struct ocfs2_lock_res_ops *ops;
545 
546 	switch(type) {
547 		case OCFS2_LOCK_TYPE_RW:
548 			ops = &ocfs2_inode_rw_lops;
549 			break;
550 		case OCFS2_LOCK_TYPE_META:
551 			ops = &ocfs2_inode_inode_lops;
552 			break;
553 		case OCFS2_LOCK_TYPE_OPEN:
554 			ops = &ocfs2_inode_open_lops;
555 			break;
556 		default:
557 			mlog_bug_on_msg(1, "type: %d\n", type);
558 			ops = NULL; /* thanks, gcc */
559 			break;
560 	};
561 
562 	ocfs2_build_lock_name(type, OCFS2_I(inode)->ip_blkno,
563 			      generation, res->l_name);
564 	ocfs2_lock_res_init_common(OCFS2_SB(inode->i_sb), res, type, ops, inode);
565 }
566 
567 static struct ocfs2_super *ocfs2_get_inode_osb(struct ocfs2_lock_res *lockres)
568 {
569 	struct inode *inode = ocfs2_lock_res_inode(lockres);
570 
571 	return OCFS2_SB(inode->i_sb);
572 }
573 
574 static struct ocfs2_super *ocfs2_get_qinfo_osb(struct ocfs2_lock_res *lockres)
575 {
576 	struct ocfs2_mem_dqinfo *info = lockres->l_priv;
577 
578 	return OCFS2_SB(info->dqi_gi.dqi_sb);
579 }
580 
581 static struct ocfs2_super *ocfs2_get_file_osb(struct ocfs2_lock_res *lockres)
582 {
583 	struct ocfs2_file_private *fp = lockres->l_priv;
584 
585 	return OCFS2_SB(fp->fp_file->f_mapping->host->i_sb);
586 }
587 
588 static __u64 ocfs2_get_dentry_lock_ino(struct ocfs2_lock_res *lockres)
589 {
590 	__be64 inode_blkno_be;
591 
592 	memcpy(&inode_blkno_be, &lockres->l_name[OCFS2_DENTRY_LOCK_INO_START],
593 	       sizeof(__be64));
594 
595 	return be64_to_cpu(inode_blkno_be);
596 }
597 
598 static struct ocfs2_super *ocfs2_get_dentry_osb(struct ocfs2_lock_res *lockres)
599 {
600 	struct ocfs2_dentry_lock *dl = lockres->l_priv;
601 
602 	return OCFS2_SB(dl->dl_inode->i_sb);
603 }
604 
605 void ocfs2_dentry_lock_res_init(struct ocfs2_dentry_lock *dl,
606 				u64 parent, struct inode *inode)
607 {
608 	int len;
609 	u64 inode_blkno = OCFS2_I(inode)->ip_blkno;
610 	__be64 inode_blkno_be = cpu_to_be64(inode_blkno);
611 	struct ocfs2_lock_res *lockres = &dl->dl_lockres;
612 
613 	ocfs2_lock_res_init_once(lockres);
614 
615 	/*
616 	 * Unfortunately, the standard lock naming scheme won't work
617 	 * here because we have two 16 byte values to use. Instead,
618 	 * we'll stuff the inode number as a binary value. We still
619 	 * want error prints to show something without garbling the
620 	 * display, so drop a null byte in there before the inode
621 	 * number. A future version of OCFS2 will likely use all
622 	 * binary lock names. The stringified names have been a
623 	 * tremendous aid in debugging, but now that the debugfs
624 	 * interface exists, we can mangle things there if need be.
625 	 *
626 	 * NOTE: We also drop the standard "pad" value (the total lock
627 	 * name size stays the same though - the last part is all
628 	 * zeros due to the memset in ocfs2_lock_res_init_once()
629 	 */
630 	len = snprintf(lockres->l_name, OCFS2_DENTRY_LOCK_INO_START,
631 		       "%c%016llx",
632 		       ocfs2_lock_type_char(OCFS2_LOCK_TYPE_DENTRY),
633 		       (long long)parent);
634 
635 	BUG_ON(len != (OCFS2_DENTRY_LOCK_INO_START - 1));
636 
637 	memcpy(&lockres->l_name[OCFS2_DENTRY_LOCK_INO_START], &inode_blkno_be,
638 	       sizeof(__be64));
639 
640 	ocfs2_lock_res_init_common(OCFS2_SB(inode->i_sb), lockres,
641 				   OCFS2_LOCK_TYPE_DENTRY, &ocfs2_dentry_lops,
642 				   dl);
643 }
644 
645 static void ocfs2_super_lock_res_init(struct ocfs2_lock_res *res,
646 				      struct ocfs2_super *osb)
647 {
648 	/* Superblock lockres doesn't come from a slab so we call init
649 	 * once on it manually.  */
650 	ocfs2_lock_res_init_once(res);
651 	ocfs2_build_lock_name(OCFS2_LOCK_TYPE_SUPER, OCFS2_SUPER_BLOCK_BLKNO,
652 			      0, res->l_name);
653 	ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_SUPER,
654 				   &ocfs2_super_lops, osb);
655 }
656 
657 static void ocfs2_rename_lock_res_init(struct ocfs2_lock_res *res,
658 				       struct ocfs2_super *osb)
659 {
660 	/* Rename 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_RENAME, 0, 0, res->l_name);
664 	ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_RENAME,
665 				   &ocfs2_rename_lops, osb);
666 }
667 
668 static void ocfs2_nfs_sync_lock_res_init(struct ocfs2_lock_res *res,
669 					 struct ocfs2_super *osb)
670 {
671 	/* nfs_sync lockres doesn't come from a slab so we call init
672 	 * once on it manually.  */
673 	ocfs2_lock_res_init_once(res);
674 	ocfs2_build_lock_name(OCFS2_LOCK_TYPE_NFS_SYNC, 0, 0, res->l_name);
675 	ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_NFS_SYNC,
676 				   &ocfs2_nfs_sync_lops, osb);
677 }
678 
679 static void ocfs2_orphan_scan_lock_res_init(struct ocfs2_lock_res *res,
680 					    struct ocfs2_super *osb)
681 {
682 	ocfs2_lock_res_init_once(res);
683 	ocfs2_build_lock_name(OCFS2_LOCK_TYPE_ORPHAN_SCAN, 0, 0, res->l_name);
684 	ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_ORPHAN_SCAN,
685 				   &ocfs2_orphan_scan_lops, osb);
686 }
687 
688 void ocfs2_file_lock_res_init(struct ocfs2_lock_res *lockres,
689 			      struct ocfs2_file_private *fp)
690 {
691 	struct inode *inode = fp->fp_file->f_mapping->host;
692 	struct ocfs2_inode_info *oi = OCFS2_I(inode);
693 
694 	ocfs2_lock_res_init_once(lockres);
695 	ocfs2_build_lock_name(OCFS2_LOCK_TYPE_FLOCK, oi->ip_blkno,
696 			      inode->i_generation, lockres->l_name);
697 	ocfs2_lock_res_init_common(OCFS2_SB(inode->i_sb), lockres,
698 				   OCFS2_LOCK_TYPE_FLOCK, &ocfs2_flock_lops,
699 				   fp);
700 	lockres->l_flags |= OCFS2_LOCK_NOCACHE;
701 }
702 
703 void ocfs2_qinfo_lock_res_init(struct ocfs2_lock_res *lockres,
704 			       struct ocfs2_mem_dqinfo *info)
705 {
706 	ocfs2_lock_res_init_once(lockres);
707 	ocfs2_build_lock_name(OCFS2_LOCK_TYPE_QINFO, info->dqi_gi.dqi_type,
708 			      0, lockres->l_name);
709 	ocfs2_lock_res_init_common(OCFS2_SB(info->dqi_gi.dqi_sb), lockres,
710 				   OCFS2_LOCK_TYPE_QINFO, &ocfs2_qinfo_lops,
711 				   info);
712 }
713 
714 void ocfs2_refcount_lock_res_init(struct ocfs2_lock_res *lockres,
715 				  struct ocfs2_super *osb, u64 ref_blkno,
716 				  unsigned int generation)
717 {
718 	ocfs2_lock_res_init_once(lockres);
719 	ocfs2_build_lock_name(OCFS2_LOCK_TYPE_REFCOUNT, ref_blkno,
720 			      generation, lockres->l_name);
721 	ocfs2_lock_res_init_common(osb, lockres, OCFS2_LOCK_TYPE_REFCOUNT,
722 				   &ocfs2_refcount_block_lops, osb);
723 }
724 
725 void ocfs2_lock_res_free(struct ocfs2_lock_res *res)
726 {
727 	if (!(res->l_flags & OCFS2_LOCK_INITIALIZED))
728 		return;
729 
730 	ocfs2_remove_lockres_tracking(res);
731 
732 	mlog_bug_on_msg(!list_empty(&res->l_blocked_list),
733 			"Lockres %s is on the blocked list\n",
734 			res->l_name);
735 	mlog_bug_on_msg(!list_empty(&res->l_mask_waiters),
736 			"Lockres %s has mask waiters pending\n",
737 			res->l_name);
738 	mlog_bug_on_msg(spin_is_locked(&res->l_lock),
739 			"Lockres %s is locked\n",
740 			res->l_name);
741 	mlog_bug_on_msg(res->l_ro_holders,
742 			"Lockres %s has %u ro holders\n",
743 			res->l_name, res->l_ro_holders);
744 	mlog_bug_on_msg(res->l_ex_holders,
745 			"Lockres %s has %u ex holders\n",
746 			res->l_name, res->l_ex_holders);
747 
748 	/* Need to clear out the lock status block for the dlm */
749 	memset(&res->l_lksb, 0, sizeof(res->l_lksb));
750 
751 	res->l_flags = 0UL;
752 }
753 
754 /*
755  * Keep a list of processes who have interest in a lockres.
756  * Note: this is now only uesed for check recursive cluster locking.
757  */
758 static inline void ocfs2_add_holder(struct ocfs2_lock_res *lockres,
759 				   struct ocfs2_lock_holder *oh)
760 {
761 	INIT_LIST_HEAD(&oh->oh_list);
762 	oh->oh_owner_pid = get_pid(task_pid(current));
763 
764 	spin_lock(&lockres->l_lock);
765 	list_add_tail(&oh->oh_list, &lockres->l_holders);
766 	spin_unlock(&lockres->l_lock);
767 }
768 
769 static inline void ocfs2_remove_holder(struct ocfs2_lock_res *lockres,
770 				       struct ocfs2_lock_holder *oh)
771 {
772 	spin_lock(&lockres->l_lock);
773 	list_del(&oh->oh_list);
774 	spin_unlock(&lockres->l_lock);
775 
776 	put_pid(oh->oh_owner_pid);
777 }
778 
779 static inline int ocfs2_is_locked_by_me(struct ocfs2_lock_res *lockres)
780 {
781 	struct ocfs2_lock_holder *oh;
782 	struct pid *pid;
783 
784 	/* look in the list of holders for one with the current task as owner */
785 	spin_lock(&lockres->l_lock);
786 	pid = task_pid(current);
787 	list_for_each_entry(oh, &lockres->l_holders, oh_list) {
788 		if (oh->oh_owner_pid == pid) {
789 			spin_unlock(&lockres->l_lock);
790 			return 1;
791 		}
792 	}
793 	spin_unlock(&lockres->l_lock);
794 
795 	return 0;
796 }
797 
798 static inline void ocfs2_inc_holders(struct ocfs2_lock_res *lockres,
799 				     int level)
800 {
801 	BUG_ON(!lockres);
802 
803 	switch(level) {
804 	case DLM_LOCK_EX:
805 		lockres->l_ex_holders++;
806 		break;
807 	case DLM_LOCK_PR:
808 		lockres->l_ro_holders++;
809 		break;
810 	default:
811 		BUG();
812 	}
813 }
814 
815 static inline void ocfs2_dec_holders(struct ocfs2_lock_res *lockres,
816 				     int level)
817 {
818 	BUG_ON(!lockres);
819 
820 	switch(level) {
821 	case DLM_LOCK_EX:
822 		BUG_ON(!lockres->l_ex_holders);
823 		lockres->l_ex_holders--;
824 		break;
825 	case DLM_LOCK_PR:
826 		BUG_ON(!lockres->l_ro_holders);
827 		lockres->l_ro_holders--;
828 		break;
829 	default:
830 		BUG();
831 	}
832 }
833 
834 /* WARNING: This function lives in a world where the only three lock
835  * levels are EX, PR, and NL. It *will* have to be adjusted when more
836  * lock types are added. */
837 static inline int ocfs2_highest_compat_lock_level(int level)
838 {
839 	int new_level = DLM_LOCK_EX;
840 
841 	if (level == DLM_LOCK_EX)
842 		new_level = DLM_LOCK_NL;
843 	else if (level == DLM_LOCK_PR)
844 		new_level = DLM_LOCK_PR;
845 	return new_level;
846 }
847 
848 static void lockres_set_flags(struct ocfs2_lock_res *lockres,
849 			      unsigned long newflags)
850 {
851 	struct ocfs2_mask_waiter *mw, *tmp;
852 
853  	assert_spin_locked(&lockres->l_lock);
854 
855 	lockres->l_flags = newflags;
856 
857 	list_for_each_entry_safe(mw, tmp, &lockres->l_mask_waiters, mw_item) {
858 		if ((lockres->l_flags & mw->mw_mask) != mw->mw_goal)
859 			continue;
860 
861 		list_del_init(&mw->mw_item);
862 		mw->mw_status = 0;
863 		complete(&mw->mw_complete);
864 	}
865 }
866 static void lockres_or_flags(struct ocfs2_lock_res *lockres, unsigned long or)
867 {
868 	lockres_set_flags(lockres, lockres->l_flags | or);
869 }
870 static void lockres_clear_flags(struct ocfs2_lock_res *lockres,
871 				unsigned long clear)
872 {
873 	lockres_set_flags(lockres, lockres->l_flags & ~clear);
874 }
875 
876 static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res *lockres)
877 {
878 	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
879 	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_ATTACHED));
880 	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BLOCKED));
881 	BUG_ON(lockres->l_blocking <= DLM_LOCK_NL);
882 
883 	lockres->l_level = lockres->l_requested;
884 	if (lockres->l_level <=
885 	    ocfs2_highest_compat_lock_level(lockres->l_blocking)) {
886 		lockres->l_blocking = DLM_LOCK_NL;
887 		lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED);
888 	}
889 	lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
890 }
891 
892 static inline void ocfs2_generic_handle_convert_action(struct ocfs2_lock_res *lockres)
893 {
894 	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
895 	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_ATTACHED));
896 
897 	/* Convert from RO to EX doesn't really need anything as our
898 	 * information is already up to data. Convert from NL to
899 	 * *anything* however should mark ourselves as needing an
900 	 * update */
901 	if (lockres->l_level == DLM_LOCK_NL &&
902 	    lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
903 		lockres_or_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
904 
905 	lockres->l_level = lockres->l_requested;
906 
907 	/*
908 	 * We set the OCFS2_LOCK_UPCONVERT_FINISHING flag before clearing
909 	 * the OCFS2_LOCK_BUSY flag to prevent the dc thread from
910 	 * downconverting the lock before the upconvert has fully completed.
911 	 * Do not prevent the dc thread from downconverting if NONBLOCK lock
912 	 * had already returned.
913 	 */
914 	if (!(lockres->l_flags & OCFS2_LOCK_NONBLOCK_FINISHED))
915 		lockres_or_flags(lockres, OCFS2_LOCK_UPCONVERT_FINISHING);
916 	else
917 		lockres_clear_flags(lockres, OCFS2_LOCK_NONBLOCK_FINISHED);
918 
919 	lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
920 }
921 
922 static inline void ocfs2_generic_handle_attach_action(struct ocfs2_lock_res *lockres)
923 {
924 	BUG_ON((!(lockres->l_flags & OCFS2_LOCK_BUSY)));
925 	BUG_ON(lockres->l_flags & OCFS2_LOCK_ATTACHED);
926 
927 	if (lockres->l_requested > DLM_LOCK_NL &&
928 	    !(lockres->l_flags & OCFS2_LOCK_LOCAL) &&
929 	    lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
930 		lockres_or_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
931 
932 	lockres->l_level = lockres->l_requested;
933 	lockres_or_flags(lockres, OCFS2_LOCK_ATTACHED);
934 	lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
935 }
936 
937 static int ocfs2_generic_handle_bast(struct ocfs2_lock_res *lockres,
938 				     int level)
939 {
940 	int needs_downconvert = 0;
941 
942 	assert_spin_locked(&lockres->l_lock);
943 
944 	if (level > lockres->l_blocking) {
945 		/* only schedule a downconvert if we haven't already scheduled
946 		 * one that goes low enough to satisfy the level we're
947 		 * blocking.  this also catches the case where we get
948 		 * duplicate BASTs */
949 		if (ocfs2_highest_compat_lock_level(level) <
950 		    ocfs2_highest_compat_lock_level(lockres->l_blocking))
951 			needs_downconvert = 1;
952 
953 		lockres->l_blocking = level;
954 	}
955 
956 	mlog(ML_BASTS, "lockres %s, block %d, level %d, l_block %d, dwn %d\n",
957 	     lockres->l_name, level, lockres->l_level, lockres->l_blocking,
958 	     needs_downconvert);
959 
960 	if (needs_downconvert)
961 		lockres_or_flags(lockres, OCFS2_LOCK_BLOCKED);
962 	mlog(0, "needs_downconvert = %d\n", needs_downconvert);
963 	return needs_downconvert;
964 }
965 
966 /*
967  * OCFS2_LOCK_PENDING and l_pending_gen.
968  *
969  * Why does OCFS2_LOCK_PENDING exist?  To close a race between setting
970  * OCFS2_LOCK_BUSY and calling ocfs2_dlm_lock().  See ocfs2_unblock_lock()
971  * for more details on the race.
972  *
973  * OCFS2_LOCK_PENDING closes the race quite nicely.  However, it introduces
974  * a race on itself.  In o2dlm, we can get the ast before ocfs2_dlm_lock()
975  * returns.  The ast clears OCFS2_LOCK_BUSY, and must therefore clear
976  * OCFS2_LOCK_PENDING at the same time.  When ocfs2_dlm_lock() returns,
977  * the caller is going to try to clear PENDING again.  If nothing else is
978  * happening, __lockres_clear_pending() sees PENDING is unset and does
979  * nothing.
980  *
981  * But what if another path (eg downconvert thread) has just started a
982  * new locking action?  The other path has re-set PENDING.  Our path
983  * cannot clear PENDING, because that will re-open the original race
984  * window.
985  *
986  * [Example]
987  *
988  * ocfs2_meta_lock()
989  *  ocfs2_cluster_lock()
990  *   set BUSY
991  *   set PENDING
992  *   drop l_lock
993  *   ocfs2_dlm_lock()
994  *    ocfs2_locking_ast()		ocfs2_downconvert_thread()
995  *     clear PENDING			 ocfs2_unblock_lock()
996  *					  take_l_lock
997  *					  !BUSY
998  *					  ocfs2_prepare_downconvert()
999  *					   set BUSY
1000  *					   set PENDING
1001  *					  drop l_lock
1002  *   take l_lock
1003  *   clear PENDING
1004  *   drop l_lock
1005  *			<window>
1006  *					  ocfs2_dlm_lock()
1007  *
1008  * So as you can see, we now have a window where l_lock is not held,
1009  * PENDING is not set, and ocfs2_dlm_lock() has not been called.
1010  *
1011  * The core problem is that ocfs2_cluster_lock() has cleared the PENDING
1012  * set by ocfs2_prepare_downconvert().  That wasn't nice.
1013  *
1014  * To solve this we introduce l_pending_gen.  A call to
1015  * lockres_clear_pending() will only do so when it is passed a generation
1016  * number that matches the lockres.  lockres_set_pending() will return the
1017  * current generation number.  When ocfs2_cluster_lock() goes to clear
1018  * PENDING, it passes the generation it got from set_pending().  In our
1019  * example above, the generation numbers will *not* match.  Thus,
1020  * ocfs2_cluster_lock() will not clear the PENDING set by
1021  * ocfs2_prepare_downconvert().
1022  */
1023 
1024 /* Unlocked version for ocfs2_locking_ast() */
1025 static void __lockres_clear_pending(struct ocfs2_lock_res *lockres,
1026 				    unsigned int generation,
1027 				    struct ocfs2_super *osb)
1028 {
1029 	assert_spin_locked(&lockres->l_lock);
1030 
1031 	/*
1032 	 * The ast and locking functions can race us here.  The winner
1033 	 * will clear pending, the loser will not.
1034 	 */
1035 	if (!(lockres->l_flags & OCFS2_LOCK_PENDING) ||
1036 	    (lockres->l_pending_gen != generation))
1037 		return;
1038 
1039 	lockres_clear_flags(lockres, OCFS2_LOCK_PENDING);
1040 	lockres->l_pending_gen++;
1041 
1042 	/*
1043 	 * The downconvert thread may have skipped us because we
1044 	 * were PENDING.  Wake it up.
1045 	 */
1046 	if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
1047 		ocfs2_wake_downconvert_thread(osb);
1048 }
1049 
1050 /* Locked version for callers of ocfs2_dlm_lock() */
1051 static void lockres_clear_pending(struct ocfs2_lock_res *lockres,
1052 				  unsigned int generation,
1053 				  struct ocfs2_super *osb)
1054 {
1055 	unsigned long flags;
1056 
1057 	spin_lock_irqsave(&lockres->l_lock, flags);
1058 	__lockres_clear_pending(lockres, generation, osb);
1059 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1060 }
1061 
1062 static unsigned int lockres_set_pending(struct ocfs2_lock_res *lockres)
1063 {
1064 	assert_spin_locked(&lockres->l_lock);
1065 	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
1066 
1067 	lockres_or_flags(lockres, OCFS2_LOCK_PENDING);
1068 
1069 	return lockres->l_pending_gen;
1070 }
1071 
1072 static void ocfs2_blocking_ast(struct ocfs2_dlm_lksb *lksb, int level)
1073 {
1074 	struct ocfs2_lock_res *lockres = ocfs2_lksb_to_lock_res(lksb);
1075 	struct ocfs2_super *osb = ocfs2_get_lockres_osb(lockres);
1076 	int needs_downconvert;
1077 	unsigned long flags;
1078 
1079 	BUG_ON(level <= DLM_LOCK_NL);
1080 
1081 	mlog(ML_BASTS, "BAST fired for lockres %s, blocking %d, level %d, "
1082 	     "type %s\n", lockres->l_name, level, lockres->l_level,
1083 	     ocfs2_lock_type_string(lockres->l_type));
1084 
1085 	/*
1086 	 * We can skip the bast for locks which don't enable caching -
1087 	 * they'll be dropped at the earliest possible time anyway.
1088 	 */
1089 	if (lockres->l_flags & OCFS2_LOCK_NOCACHE)
1090 		return;
1091 
1092 	spin_lock_irqsave(&lockres->l_lock, flags);
1093 	needs_downconvert = ocfs2_generic_handle_bast(lockres, level);
1094 	if (needs_downconvert)
1095 		ocfs2_schedule_blocked_lock(osb, lockres);
1096 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1097 
1098 	wake_up(&lockres->l_event);
1099 
1100 	ocfs2_wake_downconvert_thread(osb);
1101 }
1102 
1103 static void ocfs2_locking_ast(struct ocfs2_dlm_lksb *lksb)
1104 {
1105 	struct ocfs2_lock_res *lockres = ocfs2_lksb_to_lock_res(lksb);
1106 	struct ocfs2_super *osb = ocfs2_get_lockres_osb(lockres);
1107 	unsigned long flags;
1108 	int status;
1109 
1110 	spin_lock_irqsave(&lockres->l_lock, flags);
1111 
1112 	status = ocfs2_dlm_lock_status(&lockres->l_lksb);
1113 
1114 	if (status == -EAGAIN) {
1115 		lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
1116 		goto out;
1117 	}
1118 
1119 	if (status) {
1120 		mlog(ML_ERROR, "lockres %s: lksb status value of %d!\n",
1121 		     lockres->l_name, status);
1122 		spin_unlock_irqrestore(&lockres->l_lock, flags);
1123 		return;
1124 	}
1125 
1126 	mlog(ML_BASTS, "AST fired for lockres %s, action %d, unlock %d, "
1127 	     "level %d => %d\n", lockres->l_name, lockres->l_action,
1128 	     lockres->l_unlock_action, lockres->l_level, lockres->l_requested);
1129 
1130 	switch(lockres->l_action) {
1131 	case OCFS2_AST_ATTACH:
1132 		ocfs2_generic_handle_attach_action(lockres);
1133 		lockres_clear_flags(lockres, OCFS2_LOCK_LOCAL);
1134 		break;
1135 	case OCFS2_AST_CONVERT:
1136 		ocfs2_generic_handle_convert_action(lockres);
1137 		break;
1138 	case OCFS2_AST_DOWNCONVERT:
1139 		ocfs2_generic_handle_downconvert_action(lockres);
1140 		break;
1141 	default:
1142 		mlog(ML_ERROR, "lockres %s: AST fired with invalid action: %u, "
1143 		     "flags 0x%lx, unlock: %u\n",
1144 		     lockres->l_name, lockres->l_action, lockres->l_flags,
1145 		     lockres->l_unlock_action);
1146 		BUG();
1147 	}
1148 out:
1149 	/* set it to something invalid so if we get called again we
1150 	 * can catch it. */
1151 	lockres->l_action = OCFS2_AST_INVALID;
1152 
1153 	/* Did we try to cancel this lock?  Clear that state */
1154 	if (lockres->l_unlock_action == OCFS2_UNLOCK_CANCEL_CONVERT)
1155 		lockres->l_unlock_action = OCFS2_UNLOCK_INVALID;
1156 
1157 	/*
1158 	 * We may have beaten the locking functions here.  We certainly
1159 	 * know that dlm_lock() has been called :-)
1160 	 * Because we can't have two lock calls in flight at once, we
1161 	 * can use lockres->l_pending_gen.
1162 	 */
1163 	__lockres_clear_pending(lockres, lockres->l_pending_gen,  osb);
1164 
1165 	wake_up(&lockres->l_event);
1166 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1167 }
1168 
1169 static void ocfs2_unlock_ast(struct ocfs2_dlm_lksb *lksb, int error)
1170 {
1171 	struct ocfs2_lock_res *lockres = ocfs2_lksb_to_lock_res(lksb);
1172 	unsigned long flags;
1173 
1174 	mlog(ML_BASTS, "UNLOCK AST fired for lockres %s, action = %d\n",
1175 	     lockres->l_name, lockres->l_unlock_action);
1176 
1177 	spin_lock_irqsave(&lockres->l_lock, flags);
1178 	if (error) {
1179 		mlog(ML_ERROR, "Dlm passes error %d for lock %s, "
1180 		     "unlock_action %d\n", error, lockres->l_name,
1181 		     lockres->l_unlock_action);
1182 		spin_unlock_irqrestore(&lockres->l_lock, flags);
1183 		return;
1184 	}
1185 
1186 	switch(lockres->l_unlock_action) {
1187 	case OCFS2_UNLOCK_CANCEL_CONVERT:
1188 		mlog(0, "Cancel convert success for %s\n", lockres->l_name);
1189 		lockres->l_action = OCFS2_AST_INVALID;
1190 		/* Downconvert thread may have requeued this lock, we
1191 		 * need to wake it. */
1192 		if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
1193 			ocfs2_wake_downconvert_thread(ocfs2_get_lockres_osb(lockres));
1194 		break;
1195 	case OCFS2_UNLOCK_DROP_LOCK:
1196 		lockres->l_level = DLM_LOCK_IV;
1197 		break;
1198 	default:
1199 		BUG();
1200 	}
1201 
1202 	lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
1203 	lockres->l_unlock_action = OCFS2_UNLOCK_INVALID;
1204 	wake_up(&lockres->l_event);
1205 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1206 }
1207 
1208 /*
1209  * This is the filesystem locking protocol.  It provides the lock handling
1210  * hooks for the underlying DLM.  It has a maximum version number.
1211  * The version number allows interoperability with systems running at
1212  * the same major number and an equal or smaller minor number.
1213  *
1214  * Whenever the filesystem does new things with locks (adds or removes a
1215  * lock, orders them differently, does different things underneath a lock),
1216  * the version must be changed.  The protocol is negotiated when joining
1217  * the dlm domain.  A node may join the domain if its major version is
1218  * identical to all other nodes and its minor version is greater than
1219  * or equal to all other nodes.  When its minor version is greater than
1220  * the other nodes, it will run at the minor version specified by the
1221  * other nodes.
1222  *
1223  * If a locking change is made that will not be compatible with older
1224  * versions, the major number must be increased and the minor version set
1225  * to zero.  If a change merely adds a behavior that can be disabled when
1226  * speaking to older versions, the minor version must be increased.  If a
1227  * change adds a fully backwards compatible change (eg, LVB changes that
1228  * are just ignored by older versions), the version does not need to be
1229  * updated.
1230  */
1231 static struct ocfs2_locking_protocol lproto = {
1232 	.lp_max_version = {
1233 		.pv_major = OCFS2_LOCKING_PROTOCOL_MAJOR,
1234 		.pv_minor = OCFS2_LOCKING_PROTOCOL_MINOR,
1235 	},
1236 	.lp_lock_ast		= ocfs2_locking_ast,
1237 	.lp_blocking_ast	= ocfs2_blocking_ast,
1238 	.lp_unlock_ast		= ocfs2_unlock_ast,
1239 };
1240 
1241 void ocfs2_set_locking_protocol(void)
1242 {
1243 	ocfs2_stack_glue_set_max_proto_version(&lproto.lp_max_version);
1244 }
1245 
1246 static inline void ocfs2_recover_from_dlm_error(struct ocfs2_lock_res *lockres,
1247 						int convert)
1248 {
1249 	unsigned long flags;
1250 
1251 	spin_lock_irqsave(&lockres->l_lock, flags);
1252 	lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
1253 	lockres_clear_flags(lockres, OCFS2_LOCK_UPCONVERT_FINISHING);
1254 	if (convert)
1255 		lockres->l_action = OCFS2_AST_INVALID;
1256 	else
1257 		lockres->l_unlock_action = OCFS2_UNLOCK_INVALID;
1258 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1259 
1260 	wake_up(&lockres->l_event);
1261 }
1262 
1263 /* Note: If we detect another process working on the lock (i.e.,
1264  * OCFS2_LOCK_BUSY), we'll bail out returning 0. It's up to the caller
1265  * to do the right thing in that case.
1266  */
1267 static int ocfs2_lock_create(struct ocfs2_super *osb,
1268 			     struct ocfs2_lock_res *lockres,
1269 			     int level,
1270 			     u32 dlm_flags)
1271 {
1272 	int ret = 0;
1273 	unsigned long flags;
1274 	unsigned int gen;
1275 
1276 	mlog(0, "lock %s, level = %d, flags = %u\n", lockres->l_name, level,
1277 	     dlm_flags);
1278 
1279 	spin_lock_irqsave(&lockres->l_lock, flags);
1280 	if ((lockres->l_flags & OCFS2_LOCK_ATTACHED) ||
1281 	    (lockres->l_flags & OCFS2_LOCK_BUSY)) {
1282 		spin_unlock_irqrestore(&lockres->l_lock, flags);
1283 		goto bail;
1284 	}
1285 
1286 	lockres->l_action = OCFS2_AST_ATTACH;
1287 	lockres->l_requested = level;
1288 	lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
1289 	gen = lockres_set_pending(lockres);
1290 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1291 
1292 	ret = ocfs2_dlm_lock(osb->cconn,
1293 			     level,
1294 			     &lockres->l_lksb,
1295 			     dlm_flags,
1296 			     lockres->l_name,
1297 			     OCFS2_LOCK_ID_MAX_LEN - 1);
1298 	lockres_clear_pending(lockres, gen, osb);
1299 	if (ret) {
1300 		ocfs2_log_dlm_error("ocfs2_dlm_lock", ret, lockres);
1301 		ocfs2_recover_from_dlm_error(lockres, 1);
1302 	}
1303 
1304 	mlog(0, "lock %s, return from ocfs2_dlm_lock\n", lockres->l_name);
1305 
1306 bail:
1307 	return ret;
1308 }
1309 
1310 static inline int ocfs2_check_wait_flag(struct ocfs2_lock_res *lockres,
1311 					int flag)
1312 {
1313 	unsigned long flags;
1314 	int ret;
1315 
1316 	spin_lock_irqsave(&lockres->l_lock, flags);
1317 	ret = lockres->l_flags & flag;
1318 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1319 
1320 	return ret;
1321 }
1322 
1323 static inline void ocfs2_wait_on_busy_lock(struct ocfs2_lock_res *lockres)
1324 
1325 {
1326 	wait_event(lockres->l_event,
1327 		   !ocfs2_check_wait_flag(lockres, OCFS2_LOCK_BUSY));
1328 }
1329 
1330 static inline void ocfs2_wait_on_refreshing_lock(struct ocfs2_lock_res *lockres)
1331 
1332 {
1333 	wait_event(lockres->l_event,
1334 		   !ocfs2_check_wait_flag(lockres, OCFS2_LOCK_REFRESHING));
1335 }
1336 
1337 /* predict what lock level we'll be dropping down to on behalf
1338  * of another node, and return true if the currently wanted
1339  * level will be compatible with it. */
1340 static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lockres,
1341 						     int wanted)
1342 {
1343 	BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BLOCKED));
1344 
1345 	return wanted <= ocfs2_highest_compat_lock_level(lockres->l_blocking);
1346 }
1347 
1348 static void ocfs2_init_mask_waiter(struct ocfs2_mask_waiter *mw)
1349 {
1350 	INIT_LIST_HEAD(&mw->mw_item);
1351 	init_completion(&mw->mw_complete);
1352 	ocfs2_init_start_time(mw);
1353 }
1354 
1355 static int ocfs2_wait_for_mask(struct ocfs2_mask_waiter *mw)
1356 {
1357 	wait_for_completion(&mw->mw_complete);
1358 	/* Re-arm the completion in case we want to wait on it again */
1359 	reinit_completion(&mw->mw_complete);
1360 	return mw->mw_status;
1361 }
1362 
1363 static void lockres_add_mask_waiter(struct ocfs2_lock_res *lockres,
1364 				    struct ocfs2_mask_waiter *mw,
1365 				    unsigned long mask,
1366 				    unsigned long goal)
1367 {
1368 	BUG_ON(!list_empty(&mw->mw_item));
1369 
1370 	assert_spin_locked(&lockres->l_lock);
1371 
1372 	list_add_tail(&mw->mw_item, &lockres->l_mask_waiters);
1373 	mw->mw_mask = mask;
1374 	mw->mw_goal = goal;
1375 }
1376 
1377 /* returns 0 if the mw that was removed was already satisfied, -EBUSY
1378  * if the mask still hadn't reached its goal */
1379 static int __lockres_remove_mask_waiter(struct ocfs2_lock_res *lockres,
1380 				      struct ocfs2_mask_waiter *mw)
1381 {
1382 	int ret = 0;
1383 
1384 	assert_spin_locked(&lockres->l_lock);
1385 	if (!list_empty(&mw->mw_item)) {
1386 		if ((lockres->l_flags & mw->mw_mask) != mw->mw_goal)
1387 			ret = -EBUSY;
1388 
1389 		list_del_init(&mw->mw_item);
1390 		init_completion(&mw->mw_complete);
1391 	}
1392 
1393 	return ret;
1394 }
1395 
1396 static int lockres_remove_mask_waiter(struct ocfs2_lock_res *lockres,
1397 				      struct ocfs2_mask_waiter *mw)
1398 {
1399 	unsigned long flags;
1400 	int ret = 0;
1401 
1402 	spin_lock_irqsave(&lockres->l_lock, flags);
1403 	ret = __lockres_remove_mask_waiter(lockres, mw);
1404 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1405 
1406 	return ret;
1407 
1408 }
1409 
1410 static int ocfs2_wait_for_mask_interruptible(struct ocfs2_mask_waiter *mw,
1411 					     struct ocfs2_lock_res *lockres)
1412 {
1413 	int ret;
1414 
1415 	ret = wait_for_completion_interruptible(&mw->mw_complete);
1416 	if (ret)
1417 		lockres_remove_mask_waiter(lockres, mw);
1418 	else
1419 		ret = mw->mw_status;
1420 	/* Re-arm the completion in case we want to wait on it again */
1421 	reinit_completion(&mw->mw_complete);
1422 	return ret;
1423 }
1424 
1425 static int __ocfs2_cluster_lock(struct ocfs2_super *osb,
1426 				struct ocfs2_lock_res *lockres,
1427 				int level,
1428 				u32 lkm_flags,
1429 				int arg_flags,
1430 				int l_subclass,
1431 				unsigned long caller_ip)
1432 {
1433 	struct ocfs2_mask_waiter mw;
1434 	int wait, catch_signals = !(osb->s_mount_opt & OCFS2_MOUNT_NOINTR);
1435 	int ret = 0; /* gcc doesn't realize wait = 1 guarantees ret is set */
1436 	unsigned long flags;
1437 	unsigned int gen;
1438 	int noqueue_attempted = 0;
1439 	int dlm_locked = 0;
1440 	int kick_dc = 0;
1441 
1442 	if (!(lockres->l_flags & OCFS2_LOCK_INITIALIZED)) {
1443 		mlog_errno(-EINVAL);
1444 		return -EINVAL;
1445 	}
1446 
1447 	ocfs2_init_mask_waiter(&mw);
1448 
1449 	if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
1450 		lkm_flags |= DLM_LKF_VALBLK;
1451 
1452 again:
1453 	wait = 0;
1454 
1455 	spin_lock_irqsave(&lockres->l_lock, flags);
1456 
1457 	if (catch_signals && signal_pending(current)) {
1458 		ret = -ERESTARTSYS;
1459 		goto unlock;
1460 	}
1461 
1462 	mlog_bug_on_msg(lockres->l_flags & OCFS2_LOCK_FREEING,
1463 			"Cluster lock called on freeing lockres %s! flags "
1464 			"0x%lx\n", lockres->l_name, lockres->l_flags);
1465 
1466 	/* We only compare against the currently granted level
1467 	 * here. If the lock is blocked waiting on a downconvert,
1468 	 * we'll get caught below. */
1469 	if (lockres->l_flags & OCFS2_LOCK_BUSY &&
1470 	    level > lockres->l_level) {
1471 		/* is someone sitting in dlm_lock? If so, wait on
1472 		 * them. */
1473 		lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1474 		wait = 1;
1475 		goto unlock;
1476 	}
1477 
1478 	if (lockres->l_flags & OCFS2_LOCK_UPCONVERT_FINISHING) {
1479 		/*
1480 		 * We've upconverted. If the lock now has a level we can
1481 		 * work with, we take it. If, however, the lock is not at the
1482 		 * required level, we go thru the full cycle. One way this could
1483 		 * happen is if a process requesting an upconvert to PR is
1484 		 * closely followed by another requesting upconvert to an EX.
1485 		 * If the process requesting EX lands here, we want it to
1486 		 * continue attempting to upconvert and let the process
1487 		 * requesting PR take the lock.
1488 		 * If multiple processes request upconvert to PR, the first one
1489 		 * here will take the lock. The others will have to go thru the
1490 		 * OCFS2_LOCK_BLOCKED check to ensure that there is no pending
1491 		 * downconvert request.
1492 		 */
1493 		if (level <= lockres->l_level)
1494 			goto update_holders;
1495 	}
1496 
1497 	if (lockres->l_flags & OCFS2_LOCK_BLOCKED &&
1498 	    !ocfs2_may_continue_on_blocked_lock(lockres, level)) {
1499 		/* is the lock is currently blocked on behalf of
1500 		 * another node */
1501 		lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BLOCKED, 0);
1502 		wait = 1;
1503 		goto unlock;
1504 	}
1505 
1506 	if (level > lockres->l_level) {
1507 		if (noqueue_attempted > 0) {
1508 			ret = -EAGAIN;
1509 			goto unlock;
1510 		}
1511 		if (lkm_flags & DLM_LKF_NOQUEUE)
1512 			noqueue_attempted = 1;
1513 
1514 		if (lockres->l_action != OCFS2_AST_INVALID)
1515 			mlog(ML_ERROR, "lockres %s has action %u pending\n",
1516 			     lockres->l_name, lockres->l_action);
1517 
1518 		if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
1519 			lockres->l_action = OCFS2_AST_ATTACH;
1520 			lkm_flags &= ~DLM_LKF_CONVERT;
1521 		} else {
1522 			lockres->l_action = OCFS2_AST_CONVERT;
1523 			lkm_flags |= DLM_LKF_CONVERT;
1524 		}
1525 
1526 		lockres->l_requested = level;
1527 		lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
1528 		gen = lockres_set_pending(lockres);
1529 		spin_unlock_irqrestore(&lockres->l_lock, flags);
1530 
1531 		BUG_ON(level == DLM_LOCK_IV);
1532 		BUG_ON(level == DLM_LOCK_NL);
1533 
1534 		mlog(ML_BASTS, "lockres %s, convert from %d to %d\n",
1535 		     lockres->l_name, lockres->l_level, level);
1536 
1537 		/* call dlm_lock to upgrade lock now */
1538 		ret = ocfs2_dlm_lock(osb->cconn,
1539 				     level,
1540 				     &lockres->l_lksb,
1541 				     lkm_flags,
1542 				     lockres->l_name,
1543 				     OCFS2_LOCK_ID_MAX_LEN - 1);
1544 		lockres_clear_pending(lockres, gen, osb);
1545 		if (ret) {
1546 			if (!(lkm_flags & DLM_LKF_NOQUEUE) ||
1547 			    (ret != -EAGAIN)) {
1548 				ocfs2_log_dlm_error("ocfs2_dlm_lock",
1549 						    ret, lockres);
1550 			}
1551 			ocfs2_recover_from_dlm_error(lockres, 1);
1552 			goto out;
1553 		}
1554 		dlm_locked = 1;
1555 
1556 		mlog(0, "lock %s, successful return from ocfs2_dlm_lock\n",
1557 		     lockres->l_name);
1558 
1559 		/* At this point we've gone inside the dlm and need to
1560 		 * complete our work regardless. */
1561 		catch_signals = 0;
1562 
1563 		/* wait for busy to clear and carry on */
1564 		goto again;
1565 	}
1566 
1567 update_holders:
1568 	/* Ok, if we get here then we're good to go. */
1569 	ocfs2_inc_holders(lockres, level);
1570 
1571 	ret = 0;
1572 unlock:
1573 	lockres_clear_flags(lockres, OCFS2_LOCK_UPCONVERT_FINISHING);
1574 
1575 	/* ocfs2_unblock_lock reques on seeing OCFS2_LOCK_UPCONVERT_FINISHING */
1576 	kick_dc = (lockres->l_flags & OCFS2_LOCK_BLOCKED);
1577 
1578 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1579 	if (kick_dc)
1580 		ocfs2_wake_downconvert_thread(osb);
1581 out:
1582 	/*
1583 	 * This is helping work around a lock inversion between the page lock
1584 	 * and dlm locks.  One path holds the page lock while calling aops
1585 	 * which block acquiring dlm locks.  The voting thread holds dlm
1586 	 * locks while acquiring page locks while down converting data locks.
1587 	 * This block is helping an aop path notice the inversion and back
1588 	 * off to unlock its page lock before trying the dlm lock again.
1589 	 */
1590 	if (wait && arg_flags & OCFS2_LOCK_NONBLOCK &&
1591 	    mw.mw_mask & (OCFS2_LOCK_BUSY|OCFS2_LOCK_BLOCKED)) {
1592 		wait = 0;
1593 		spin_lock_irqsave(&lockres->l_lock, flags);
1594 		if (__lockres_remove_mask_waiter(lockres, &mw)) {
1595 			if (dlm_locked)
1596 				lockres_or_flags(lockres,
1597 					OCFS2_LOCK_NONBLOCK_FINISHED);
1598 			spin_unlock_irqrestore(&lockres->l_lock, flags);
1599 			ret = -EAGAIN;
1600 		} else {
1601 			spin_unlock_irqrestore(&lockres->l_lock, flags);
1602 			goto again;
1603 		}
1604 	}
1605 	if (wait) {
1606 		ret = ocfs2_wait_for_mask(&mw);
1607 		if (ret == 0)
1608 			goto again;
1609 		mlog_errno(ret);
1610 	}
1611 	ocfs2_update_lock_stats(lockres, level, &mw, ret);
1612 
1613 #ifdef CONFIG_DEBUG_LOCK_ALLOC
1614 	if (!ret && lockres->l_lockdep_map.key != NULL) {
1615 		if (level == DLM_LOCK_PR)
1616 			rwsem_acquire_read(&lockres->l_lockdep_map, l_subclass,
1617 				!!(arg_flags & OCFS2_META_LOCK_NOQUEUE),
1618 				caller_ip);
1619 		else
1620 			rwsem_acquire(&lockres->l_lockdep_map, l_subclass,
1621 				!!(arg_flags & OCFS2_META_LOCK_NOQUEUE),
1622 				caller_ip);
1623 	}
1624 #endif
1625 	return ret;
1626 }
1627 
1628 static inline int ocfs2_cluster_lock(struct ocfs2_super *osb,
1629 				     struct ocfs2_lock_res *lockres,
1630 				     int level,
1631 				     u32 lkm_flags,
1632 				     int arg_flags)
1633 {
1634 	return __ocfs2_cluster_lock(osb, lockres, level, lkm_flags, arg_flags,
1635 				    0, _RET_IP_);
1636 }
1637 
1638 
1639 static void __ocfs2_cluster_unlock(struct ocfs2_super *osb,
1640 				   struct ocfs2_lock_res *lockres,
1641 				   int level,
1642 				   unsigned long caller_ip)
1643 {
1644 	unsigned long flags;
1645 
1646 	spin_lock_irqsave(&lockres->l_lock, flags);
1647 	ocfs2_dec_holders(lockres, level);
1648 	ocfs2_downconvert_on_unlock(osb, lockres);
1649 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1650 #ifdef CONFIG_DEBUG_LOCK_ALLOC
1651 	if (lockres->l_lockdep_map.key != NULL)
1652 		rwsem_release(&lockres->l_lockdep_map, 1, caller_ip);
1653 #endif
1654 }
1655 
1656 static int ocfs2_create_new_lock(struct ocfs2_super *osb,
1657 				 struct ocfs2_lock_res *lockres,
1658 				 int ex,
1659 				 int local)
1660 {
1661 	int level =  ex ? DLM_LOCK_EX : DLM_LOCK_PR;
1662 	unsigned long flags;
1663 	u32 lkm_flags = local ? DLM_LKF_LOCAL : 0;
1664 
1665 	spin_lock_irqsave(&lockres->l_lock, flags);
1666 	BUG_ON(lockres->l_flags & OCFS2_LOCK_ATTACHED);
1667 	lockres_or_flags(lockres, OCFS2_LOCK_LOCAL);
1668 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1669 
1670 	return ocfs2_lock_create(osb, lockres, level, lkm_flags);
1671 }
1672 
1673 /* Grants us an EX lock on the data and metadata resources, skipping
1674  * the normal cluster directory lookup. Use this ONLY on newly created
1675  * inodes which other nodes can't possibly see, and which haven't been
1676  * hashed in the inode hash yet. This can give us a good performance
1677  * increase as it'll skip the network broadcast normally associated
1678  * with creating a new lock resource. */
1679 int ocfs2_create_new_inode_locks(struct inode *inode)
1680 {
1681 	int ret;
1682 	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1683 
1684 	BUG_ON(!ocfs2_inode_is_new(inode));
1685 
1686 	mlog(0, "Inode %llu\n", (unsigned long long)OCFS2_I(inode)->ip_blkno);
1687 
1688 	/* NOTE: That we don't increment any of the holder counts, nor
1689 	 * do we add anything to a journal handle. Since this is
1690 	 * supposed to be a new inode which the cluster doesn't know
1691 	 * about yet, there is no need to.  As far as the LVB handling
1692 	 * is concerned, this is basically like acquiring an EX lock
1693 	 * on a resource which has an invalid one -- we'll set it
1694 	 * valid when we release the EX. */
1695 
1696 	ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_rw_lockres, 1, 1);
1697 	if (ret) {
1698 		mlog_errno(ret);
1699 		goto bail;
1700 	}
1701 
1702 	/*
1703 	 * We don't want to use DLM_LKF_LOCAL on a meta data lock as they
1704 	 * don't use a generation in their lock names.
1705 	 */
1706 	ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_inode_lockres, 1, 0);
1707 	if (ret) {
1708 		mlog_errno(ret);
1709 		goto bail;
1710 	}
1711 
1712 	ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_open_lockres, 0, 0);
1713 	if (ret)
1714 		mlog_errno(ret);
1715 
1716 bail:
1717 	return ret;
1718 }
1719 
1720 int ocfs2_rw_lock(struct inode *inode, int write)
1721 {
1722 	int status, level;
1723 	struct ocfs2_lock_res *lockres;
1724 	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1725 
1726 	mlog(0, "inode %llu take %s RW lock\n",
1727 	     (unsigned long long)OCFS2_I(inode)->ip_blkno,
1728 	     write ? "EXMODE" : "PRMODE");
1729 
1730 	if (ocfs2_mount_local(osb))
1731 		return 0;
1732 
1733 	lockres = &OCFS2_I(inode)->ip_rw_lockres;
1734 
1735 	level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1736 
1737 	status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres, level, 0,
1738 				    0);
1739 	if (status < 0)
1740 		mlog_errno(status);
1741 
1742 	return status;
1743 }
1744 
1745 void ocfs2_rw_unlock(struct inode *inode, int write)
1746 {
1747 	int level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1748 	struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_rw_lockres;
1749 	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1750 
1751 	mlog(0, "inode %llu drop %s RW lock\n",
1752 	     (unsigned long long)OCFS2_I(inode)->ip_blkno,
1753 	     write ? "EXMODE" : "PRMODE");
1754 
1755 	if (!ocfs2_mount_local(osb))
1756 		ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres, level);
1757 }
1758 
1759 /*
1760  * ocfs2_open_lock always get PR mode lock.
1761  */
1762 int ocfs2_open_lock(struct inode *inode)
1763 {
1764 	int status = 0;
1765 	struct ocfs2_lock_res *lockres;
1766 	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1767 
1768 	mlog(0, "inode %llu take PRMODE open lock\n",
1769 	     (unsigned long long)OCFS2_I(inode)->ip_blkno);
1770 
1771 	if (ocfs2_is_hard_readonly(osb) || ocfs2_mount_local(osb))
1772 		goto out;
1773 
1774 	lockres = &OCFS2_I(inode)->ip_open_lockres;
1775 
1776 	status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres,
1777 				    DLM_LOCK_PR, 0, 0);
1778 	if (status < 0)
1779 		mlog_errno(status);
1780 
1781 out:
1782 	return status;
1783 }
1784 
1785 int ocfs2_try_open_lock(struct inode *inode, int write)
1786 {
1787 	int status = 0, level;
1788 	struct ocfs2_lock_res *lockres;
1789 	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1790 
1791 	mlog(0, "inode %llu try to take %s open lock\n",
1792 	     (unsigned long long)OCFS2_I(inode)->ip_blkno,
1793 	     write ? "EXMODE" : "PRMODE");
1794 
1795 	if (ocfs2_is_hard_readonly(osb)) {
1796 		if (write)
1797 			status = -EROFS;
1798 		goto out;
1799 	}
1800 
1801 	if (ocfs2_mount_local(osb))
1802 		goto out;
1803 
1804 	lockres = &OCFS2_I(inode)->ip_open_lockres;
1805 
1806 	level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1807 
1808 	/*
1809 	 * The file system may already holding a PRMODE/EXMODE open lock.
1810 	 * Since we pass DLM_LKF_NOQUEUE, the request won't block waiting on
1811 	 * other nodes and the -EAGAIN will indicate to the caller that
1812 	 * this inode is still in use.
1813 	 */
1814 	status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres,
1815 				    level, DLM_LKF_NOQUEUE, 0);
1816 
1817 out:
1818 	return status;
1819 }
1820 
1821 /*
1822  * ocfs2_open_unlock unlock PR and EX mode open locks.
1823  */
1824 void ocfs2_open_unlock(struct inode *inode)
1825 {
1826 	struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_open_lockres;
1827 	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1828 
1829 	mlog(0, "inode %llu drop open lock\n",
1830 	     (unsigned long long)OCFS2_I(inode)->ip_blkno);
1831 
1832 	if (ocfs2_mount_local(osb))
1833 		goto out;
1834 
1835 	if(lockres->l_ro_holders)
1836 		ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres,
1837 				     DLM_LOCK_PR);
1838 	if(lockres->l_ex_holders)
1839 		ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres,
1840 				     DLM_LOCK_EX);
1841 
1842 out:
1843 	return;
1844 }
1845 
1846 static int ocfs2_flock_handle_signal(struct ocfs2_lock_res *lockres,
1847 				     int level)
1848 {
1849 	int ret;
1850 	struct ocfs2_super *osb = ocfs2_get_lockres_osb(lockres);
1851 	unsigned long flags;
1852 	struct ocfs2_mask_waiter mw;
1853 
1854 	ocfs2_init_mask_waiter(&mw);
1855 
1856 retry_cancel:
1857 	spin_lock_irqsave(&lockres->l_lock, flags);
1858 	if (lockres->l_flags & OCFS2_LOCK_BUSY) {
1859 		ret = ocfs2_prepare_cancel_convert(osb, lockres);
1860 		if (ret) {
1861 			spin_unlock_irqrestore(&lockres->l_lock, flags);
1862 			ret = ocfs2_cancel_convert(osb, lockres);
1863 			if (ret < 0) {
1864 				mlog_errno(ret);
1865 				goto out;
1866 			}
1867 			goto retry_cancel;
1868 		}
1869 		lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1870 		spin_unlock_irqrestore(&lockres->l_lock, flags);
1871 
1872 		ocfs2_wait_for_mask(&mw);
1873 		goto retry_cancel;
1874 	}
1875 
1876 	ret = -ERESTARTSYS;
1877 	/*
1878 	 * We may still have gotten the lock, in which case there's no
1879 	 * point to restarting the syscall.
1880 	 */
1881 	if (lockres->l_level == level)
1882 		ret = 0;
1883 
1884 	mlog(0, "Cancel returning %d. flags: 0x%lx, level: %d, act: %d\n", ret,
1885 	     lockres->l_flags, lockres->l_level, lockres->l_action);
1886 
1887 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1888 
1889 out:
1890 	return ret;
1891 }
1892 
1893 /*
1894  * ocfs2_file_lock() and ocfs2_file_unlock() map to a single pair of
1895  * flock() calls. The locking approach this requires is sufficiently
1896  * different from all other cluster lock types that we implement a
1897  * separate path to the "low-level" dlm calls. In particular:
1898  *
1899  * - No optimization of lock levels is done - we take at exactly
1900  *   what's been requested.
1901  *
1902  * - No lock caching is employed. We immediately downconvert to
1903  *   no-lock at unlock time. This also means flock locks never go on
1904  *   the blocking list).
1905  *
1906  * - Since userspace can trivially deadlock itself with flock, we make
1907  *   sure to allow cancellation of a misbehaving applications flock()
1908  *   request.
1909  *
1910  * - Access to any flock lockres doesn't require concurrency, so we
1911  *   can simplify the code by requiring the caller to guarantee
1912  *   serialization of dlmglue flock calls.
1913  */
1914 int ocfs2_file_lock(struct file *file, int ex, int trylock)
1915 {
1916 	int ret, level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
1917 	unsigned int lkm_flags = trylock ? DLM_LKF_NOQUEUE : 0;
1918 	unsigned long flags;
1919 	struct ocfs2_file_private *fp = file->private_data;
1920 	struct ocfs2_lock_res *lockres = &fp->fp_flock;
1921 	struct ocfs2_super *osb = OCFS2_SB(file->f_mapping->host->i_sb);
1922 	struct ocfs2_mask_waiter mw;
1923 
1924 	ocfs2_init_mask_waiter(&mw);
1925 
1926 	if ((lockres->l_flags & OCFS2_LOCK_BUSY) ||
1927 	    (lockres->l_level > DLM_LOCK_NL)) {
1928 		mlog(ML_ERROR,
1929 		     "File lock \"%s\" has busy or locked state: flags: 0x%lx, "
1930 		     "level: %u\n", lockres->l_name, lockres->l_flags,
1931 		     lockres->l_level);
1932 		return -EINVAL;
1933 	}
1934 
1935 	spin_lock_irqsave(&lockres->l_lock, flags);
1936 	if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
1937 		lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1938 		spin_unlock_irqrestore(&lockres->l_lock, flags);
1939 
1940 		/*
1941 		 * Get the lock at NLMODE to start - that way we
1942 		 * can cancel the upconvert request if need be.
1943 		 */
1944 		ret = ocfs2_lock_create(osb, lockres, DLM_LOCK_NL, 0);
1945 		if (ret < 0) {
1946 			mlog_errno(ret);
1947 			goto out;
1948 		}
1949 
1950 		ret = ocfs2_wait_for_mask(&mw);
1951 		if (ret) {
1952 			mlog_errno(ret);
1953 			goto out;
1954 		}
1955 		spin_lock_irqsave(&lockres->l_lock, flags);
1956 	}
1957 
1958 	lockres->l_action = OCFS2_AST_CONVERT;
1959 	lkm_flags |= DLM_LKF_CONVERT;
1960 	lockres->l_requested = level;
1961 	lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
1962 
1963 	lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1964 	spin_unlock_irqrestore(&lockres->l_lock, flags);
1965 
1966 	ret = ocfs2_dlm_lock(osb->cconn, level, &lockres->l_lksb, lkm_flags,
1967 			     lockres->l_name, OCFS2_LOCK_ID_MAX_LEN - 1);
1968 	if (ret) {
1969 		if (!trylock || (ret != -EAGAIN)) {
1970 			ocfs2_log_dlm_error("ocfs2_dlm_lock", ret, lockres);
1971 			ret = -EINVAL;
1972 		}
1973 
1974 		ocfs2_recover_from_dlm_error(lockres, 1);
1975 		lockres_remove_mask_waiter(lockres, &mw);
1976 		goto out;
1977 	}
1978 
1979 	ret = ocfs2_wait_for_mask_interruptible(&mw, lockres);
1980 	if (ret == -ERESTARTSYS) {
1981 		/*
1982 		 * Userspace can cause deadlock itself with
1983 		 * flock(). Current behavior locally is to allow the
1984 		 * deadlock, but abort the system call if a signal is
1985 		 * received. We follow this example, otherwise a
1986 		 * poorly written program could sit in kernel until
1987 		 * reboot.
1988 		 *
1989 		 * Handling this is a bit more complicated for Ocfs2
1990 		 * though. We can't exit this function with an
1991 		 * outstanding lock request, so a cancel convert is
1992 		 * required. We intentionally overwrite 'ret' - if the
1993 		 * cancel fails and the lock was granted, it's easier
1994 		 * to just bubble success back up to the user.
1995 		 */
1996 		ret = ocfs2_flock_handle_signal(lockres, level);
1997 	} else if (!ret && (level > lockres->l_level)) {
1998 		/* Trylock failed asynchronously */
1999 		BUG_ON(!trylock);
2000 		ret = -EAGAIN;
2001 	}
2002 
2003 out:
2004 
2005 	mlog(0, "Lock: \"%s\" ex: %d, trylock: %d, returns: %d\n",
2006 	     lockres->l_name, ex, trylock, ret);
2007 	return ret;
2008 }
2009 
2010 void ocfs2_file_unlock(struct file *file)
2011 {
2012 	int ret;
2013 	unsigned int gen;
2014 	unsigned long flags;
2015 	struct ocfs2_file_private *fp = file->private_data;
2016 	struct ocfs2_lock_res *lockres = &fp->fp_flock;
2017 	struct ocfs2_super *osb = OCFS2_SB(file->f_mapping->host->i_sb);
2018 	struct ocfs2_mask_waiter mw;
2019 
2020 	ocfs2_init_mask_waiter(&mw);
2021 
2022 	if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED))
2023 		return;
2024 
2025 	if (lockres->l_level == DLM_LOCK_NL)
2026 		return;
2027 
2028 	mlog(0, "Unlock: \"%s\" flags: 0x%lx, level: %d, act: %d\n",
2029 	     lockres->l_name, lockres->l_flags, lockres->l_level,
2030 	     lockres->l_action);
2031 
2032 	spin_lock_irqsave(&lockres->l_lock, flags);
2033 	/*
2034 	 * Fake a blocking ast for the downconvert code.
2035 	 */
2036 	lockres_or_flags(lockres, OCFS2_LOCK_BLOCKED);
2037 	lockres->l_blocking = DLM_LOCK_EX;
2038 
2039 	gen = ocfs2_prepare_downconvert(lockres, DLM_LOCK_NL);
2040 	lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
2041 	spin_unlock_irqrestore(&lockres->l_lock, flags);
2042 
2043 	ret = ocfs2_downconvert_lock(osb, lockres, DLM_LOCK_NL, 0, gen);
2044 	if (ret) {
2045 		mlog_errno(ret);
2046 		return;
2047 	}
2048 
2049 	ret = ocfs2_wait_for_mask(&mw);
2050 	if (ret)
2051 		mlog_errno(ret);
2052 }
2053 
2054 static void ocfs2_downconvert_on_unlock(struct ocfs2_super *osb,
2055 					struct ocfs2_lock_res *lockres)
2056 {
2057 	int kick = 0;
2058 
2059 	/* If we know that another node is waiting on our lock, kick
2060 	 * the downconvert thread * pre-emptively when we reach a release
2061 	 * condition. */
2062 	if (lockres->l_flags & OCFS2_LOCK_BLOCKED) {
2063 		switch(lockres->l_blocking) {
2064 		case DLM_LOCK_EX:
2065 			if (!lockres->l_ex_holders && !lockres->l_ro_holders)
2066 				kick = 1;
2067 			break;
2068 		case DLM_LOCK_PR:
2069 			if (!lockres->l_ex_holders)
2070 				kick = 1;
2071 			break;
2072 		default:
2073 			BUG();
2074 		}
2075 	}
2076 
2077 	if (kick)
2078 		ocfs2_wake_downconvert_thread(osb);
2079 }
2080 
2081 #define OCFS2_SEC_BITS   34
2082 #define OCFS2_SEC_SHIFT  (64 - 34)
2083 #define OCFS2_NSEC_MASK  ((1ULL << OCFS2_SEC_SHIFT) - 1)
2084 
2085 /* LVB only has room for 64 bits of time here so we pack it for
2086  * now. */
2087 static u64 ocfs2_pack_timespec(struct timespec *spec)
2088 {
2089 	u64 res;
2090 	u64 sec = spec->tv_sec;
2091 	u32 nsec = spec->tv_nsec;
2092 
2093 	res = (sec << OCFS2_SEC_SHIFT) | (nsec & OCFS2_NSEC_MASK);
2094 
2095 	return res;
2096 }
2097 
2098 /* Call this with the lockres locked. I am reasonably sure we don't
2099  * need ip_lock in this function as anyone who would be changing those
2100  * values is supposed to be blocked in ocfs2_inode_lock right now. */
2101 static void __ocfs2_stuff_meta_lvb(struct inode *inode)
2102 {
2103 	struct ocfs2_inode_info *oi = OCFS2_I(inode);
2104 	struct ocfs2_lock_res *lockres = &oi->ip_inode_lockres;
2105 	struct ocfs2_meta_lvb *lvb;
2106 
2107 	lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2108 
2109 	/*
2110 	 * Invalidate the LVB of a deleted inode - this way other
2111 	 * nodes are forced to go to disk and discover the new inode
2112 	 * status.
2113 	 */
2114 	if (oi->ip_flags & OCFS2_INODE_DELETED) {
2115 		lvb->lvb_version = 0;
2116 		goto out;
2117 	}
2118 
2119 	lvb->lvb_version   = OCFS2_LVB_VERSION;
2120 	lvb->lvb_isize	   = cpu_to_be64(i_size_read(inode));
2121 	lvb->lvb_iclusters = cpu_to_be32(oi->ip_clusters);
2122 	lvb->lvb_iuid      = cpu_to_be32(i_uid_read(inode));
2123 	lvb->lvb_igid      = cpu_to_be32(i_gid_read(inode));
2124 	lvb->lvb_imode     = cpu_to_be16(inode->i_mode);
2125 	lvb->lvb_inlink    = cpu_to_be16(inode->i_nlink);
2126 	lvb->lvb_iatime_packed  =
2127 		cpu_to_be64(ocfs2_pack_timespec(&inode->i_atime));
2128 	lvb->lvb_ictime_packed =
2129 		cpu_to_be64(ocfs2_pack_timespec(&inode->i_ctime));
2130 	lvb->lvb_imtime_packed =
2131 		cpu_to_be64(ocfs2_pack_timespec(&inode->i_mtime));
2132 	lvb->lvb_iattr    = cpu_to_be32(oi->ip_attr);
2133 	lvb->lvb_idynfeatures = cpu_to_be16(oi->ip_dyn_features);
2134 	lvb->lvb_igeneration = cpu_to_be32(inode->i_generation);
2135 
2136 out:
2137 	mlog_meta_lvb(0, lockres);
2138 }
2139 
2140 static void ocfs2_unpack_timespec(struct timespec *spec,
2141 				  u64 packed_time)
2142 {
2143 	spec->tv_sec = packed_time >> OCFS2_SEC_SHIFT;
2144 	spec->tv_nsec = packed_time & OCFS2_NSEC_MASK;
2145 }
2146 
2147 static void ocfs2_refresh_inode_from_lvb(struct inode *inode)
2148 {
2149 	struct ocfs2_inode_info *oi = OCFS2_I(inode);
2150 	struct ocfs2_lock_res *lockres = &oi->ip_inode_lockres;
2151 	struct ocfs2_meta_lvb *lvb;
2152 
2153 	mlog_meta_lvb(0, lockres);
2154 
2155 	lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2156 
2157 	/* We're safe here without the lockres lock... */
2158 	spin_lock(&oi->ip_lock);
2159 	oi->ip_clusters = be32_to_cpu(lvb->lvb_iclusters);
2160 	i_size_write(inode, be64_to_cpu(lvb->lvb_isize));
2161 
2162 	oi->ip_attr = be32_to_cpu(lvb->lvb_iattr);
2163 	oi->ip_dyn_features = be16_to_cpu(lvb->lvb_idynfeatures);
2164 	ocfs2_set_inode_flags(inode);
2165 
2166 	/* fast-symlinks are a special case */
2167 	if (S_ISLNK(inode->i_mode) && !oi->ip_clusters)
2168 		inode->i_blocks = 0;
2169 	else
2170 		inode->i_blocks = ocfs2_inode_sector_count(inode);
2171 
2172 	i_uid_write(inode, be32_to_cpu(lvb->lvb_iuid));
2173 	i_gid_write(inode, be32_to_cpu(lvb->lvb_igid));
2174 	inode->i_mode    = be16_to_cpu(lvb->lvb_imode);
2175 	set_nlink(inode, be16_to_cpu(lvb->lvb_inlink));
2176 	ocfs2_unpack_timespec(&inode->i_atime,
2177 			      be64_to_cpu(lvb->lvb_iatime_packed));
2178 	ocfs2_unpack_timespec(&inode->i_mtime,
2179 			      be64_to_cpu(lvb->lvb_imtime_packed));
2180 	ocfs2_unpack_timespec(&inode->i_ctime,
2181 			      be64_to_cpu(lvb->lvb_ictime_packed));
2182 	spin_unlock(&oi->ip_lock);
2183 }
2184 
2185 static inline int ocfs2_meta_lvb_is_trustable(struct inode *inode,
2186 					      struct ocfs2_lock_res *lockres)
2187 {
2188 	struct ocfs2_meta_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2189 
2190 	if (ocfs2_dlm_lvb_valid(&lockres->l_lksb)
2191 	    && lvb->lvb_version == OCFS2_LVB_VERSION
2192 	    && be32_to_cpu(lvb->lvb_igeneration) == inode->i_generation)
2193 		return 1;
2194 	return 0;
2195 }
2196 
2197 /* Determine whether a lock resource needs to be refreshed, and
2198  * arbitrate who gets to refresh it.
2199  *
2200  *   0 means no refresh needed.
2201  *
2202  *   > 0 means you need to refresh this and you MUST call
2203  *   ocfs2_complete_lock_res_refresh afterwards. */
2204 static int ocfs2_should_refresh_lock_res(struct ocfs2_lock_res *lockres)
2205 {
2206 	unsigned long flags;
2207 	int status = 0;
2208 
2209 refresh_check:
2210 	spin_lock_irqsave(&lockres->l_lock, flags);
2211 	if (!(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH)) {
2212 		spin_unlock_irqrestore(&lockres->l_lock, flags);
2213 		goto bail;
2214 	}
2215 
2216 	if (lockres->l_flags & OCFS2_LOCK_REFRESHING) {
2217 		spin_unlock_irqrestore(&lockres->l_lock, flags);
2218 
2219 		ocfs2_wait_on_refreshing_lock(lockres);
2220 		goto refresh_check;
2221 	}
2222 
2223 	/* Ok, I'll be the one to refresh this lock. */
2224 	lockres_or_flags(lockres, OCFS2_LOCK_REFRESHING);
2225 	spin_unlock_irqrestore(&lockres->l_lock, flags);
2226 
2227 	status = 1;
2228 bail:
2229 	mlog(0, "status %d\n", status);
2230 	return status;
2231 }
2232 
2233 /* If status is non zero, I'll mark it as not being in refresh
2234  * anymroe, but i won't clear the needs refresh flag. */
2235 static inline void ocfs2_complete_lock_res_refresh(struct ocfs2_lock_res *lockres,
2236 						   int status)
2237 {
2238 	unsigned long flags;
2239 
2240 	spin_lock_irqsave(&lockres->l_lock, flags);
2241 	lockres_clear_flags(lockres, OCFS2_LOCK_REFRESHING);
2242 	if (!status)
2243 		lockres_clear_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
2244 	spin_unlock_irqrestore(&lockres->l_lock, flags);
2245 
2246 	wake_up(&lockres->l_event);
2247 }
2248 
2249 /* may or may not return a bh if it went to disk. */
2250 static int ocfs2_inode_lock_update(struct inode *inode,
2251 				  struct buffer_head **bh)
2252 {
2253 	int status = 0;
2254 	struct ocfs2_inode_info *oi = OCFS2_I(inode);
2255 	struct ocfs2_lock_res *lockres = &oi->ip_inode_lockres;
2256 	struct ocfs2_dinode *fe;
2257 	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
2258 
2259 	if (ocfs2_mount_local(osb))
2260 		goto bail;
2261 
2262 	spin_lock(&oi->ip_lock);
2263 	if (oi->ip_flags & OCFS2_INODE_DELETED) {
2264 		mlog(0, "Orphaned inode %llu was deleted while we "
2265 		     "were waiting on a lock. ip_flags = 0x%x\n",
2266 		     (unsigned long long)oi->ip_blkno, oi->ip_flags);
2267 		spin_unlock(&oi->ip_lock);
2268 		status = -ENOENT;
2269 		goto bail;
2270 	}
2271 	spin_unlock(&oi->ip_lock);
2272 
2273 	if (!ocfs2_should_refresh_lock_res(lockres))
2274 		goto bail;
2275 
2276 	/* This will discard any caching information we might have had
2277 	 * for the inode metadata. */
2278 	ocfs2_metadata_cache_purge(INODE_CACHE(inode));
2279 
2280 	ocfs2_extent_map_trunc(inode, 0);
2281 
2282 	if (ocfs2_meta_lvb_is_trustable(inode, lockres)) {
2283 		mlog(0, "Trusting LVB on inode %llu\n",
2284 		     (unsigned long long)oi->ip_blkno);
2285 		ocfs2_refresh_inode_from_lvb(inode);
2286 	} else {
2287 		/* Boo, we have to go to disk. */
2288 		/* read bh, cast, ocfs2_refresh_inode */
2289 		status = ocfs2_read_inode_block(inode, bh);
2290 		if (status < 0) {
2291 			mlog_errno(status);
2292 			goto bail_refresh;
2293 		}
2294 		fe = (struct ocfs2_dinode *) (*bh)->b_data;
2295 
2296 		/* This is a good chance to make sure we're not
2297 		 * locking an invalid object.  ocfs2_read_inode_block()
2298 		 * already checked that the inode block is sane.
2299 		 *
2300 		 * We bug on a stale inode here because we checked
2301 		 * above whether it was wiped from disk. The wiping
2302 		 * node provides a guarantee that we receive that
2303 		 * message and can mark the inode before dropping any
2304 		 * locks associated with it. */
2305 		mlog_bug_on_msg(inode->i_generation !=
2306 				le32_to_cpu(fe->i_generation),
2307 				"Invalid dinode %llu disk generation: %u "
2308 				"inode->i_generation: %u\n",
2309 				(unsigned long long)oi->ip_blkno,
2310 				le32_to_cpu(fe->i_generation),
2311 				inode->i_generation);
2312 		mlog_bug_on_msg(le64_to_cpu(fe->i_dtime) ||
2313 				!(fe->i_flags & cpu_to_le32(OCFS2_VALID_FL)),
2314 				"Stale dinode %llu dtime: %llu flags: 0x%x\n",
2315 				(unsigned long long)oi->ip_blkno,
2316 				(unsigned long long)le64_to_cpu(fe->i_dtime),
2317 				le32_to_cpu(fe->i_flags));
2318 
2319 		ocfs2_refresh_inode(inode, fe);
2320 		ocfs2_track_lock_refresh(lockres);
2321 	}
2322 
2323 	status = 0;
2324 bail_refresh:
2325 	ocfs2_complete_lock_res_refresh(lockres, status);
2326 bail:
2327 	return status;
2328 }
2329 
2330 static int ocfs2_assign_bh(struct inode *inode,
2331 			   struct buffer_head **ret_bh,
2332 			   struct buffer_head *passed_bh)
2333 {
2334 	int status;
2335 
2336 	if (passed_bh) {
2337 		/* Ok, the update went to disk for us, use the
2338 		 * returned bh. */
2339 		*ret_bh = passed_bh;
2340 		get_bh(*ret_bh);
2341 
2342 		return 0;
2343 	}
2344 
2345 	status = ocfs2_read_inode_block(inode, ret_bh);
2346 	if (status < 0)
2347 		mlog_errno(status);
2348 
2349 	return status;
2350 }
2351 
2352 /*
2353  * returns < 0 error if the callback will never be called, otherwise
2354  * the result of the lock will be communicated via the callback.
2355  */
2356 int ocfs2_inode_lock_full_nested(struct inode *inode,
2357 				 struct buffer_head **ret_bh,
2358 				 int ex,
2359 				 int arg_flags,
2360 				 int subclass)
2361 {
2362 	int status, level, acquired;
2363 	u32 dlm_flags;
2364 	struct ocfs2_lock_res *lockres = NULL;
2365 	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
2366 	struct buffer_head *local_bh = NULL;
2367 
2368 	mlog(0, "inode %llu, take %s META lock\n",
2369 	     (unsigned long long)OCFS2_I(inode)->ip_blkno,
2370 	     ex ? "EXMODE" : "PRMODE");
2371 
2372 	status = 0;
2373 	acquired = 0;
2374 	/* We'll allow faking a readonly metadata lock for
2375 	 * rodevices. */
2376 	if (ocfs2_is_hard_readonly(osb)) {
2377 		if (ex)
2378 			status = -EROFS;
2379 		goto getbh;
2380 	}
2381 
2382 	if ((arg_flags & OCFS2_META_LOCK_GETBH) ||
2383 	    ocfs2_mount_local(osb))
2384 		goto update;
2385 
2386 	if (!(arg_flags & OCFS2_META_LOCK_RECOVERY))
2387 		ocfs2_wait_for_recovery(osb);
2388 
2389 	lockres = &OCFS2_I(inode)->ip_inode_lockres;
2390 	level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2391 	dlm_flags = 0;
2392 	if (arg_flags & OCFS2_META_LOCK_NOQUEUE)
2393 		dlm_flags |= DLM_LKF_NOQUEUE;
2394 
2395 	status = __ocfs2_cluster_lock(osb, lockres, level, dlm_flags,
2396 				      arg_flags, subclass, _RET_IP_);
2397 	if (status < 0) {
2398 		if (status != -EAGAIN)
2399 			mlog_errno(status);
2400 		goto bail;
2401 	}
2402 
2403 	/* Notify the error cleanup path to drop the cluster lock. */
2404 	acquired = 1;
2405 
2406 	/* We wait twice because a node may have died while we were in
2407 	 * the lower dlm layers. The second time though, we've
2408 	 * committed to owning this lock so we don't allow signals to
2409 	 * abort the operation. */
2410 	if (!(arg_flags & OCFS2_META_LOCK_RECOVERY))
2411 		ocfs2_wait_for_recovery(osb);
2412 
2413 update:
2414 	/*
2415 	 * We only see this flag if we're being called from
2416 	 * ocfs2_read_locked_inode(). It means we're locking an inode
2417 	 * which hasn't been populated yet, so clear the refresh flag
2418 	 * and let the caller handle it.
2419 	 */
2420 	if (inode->i_state & I_NEW) {
2421 		status = 0;
2422 		if (lockres)
2423 			ocfs2_complete_lock_res_refresh(lockres, 0);
2424 		goto bail;
2425 	}
2426 
2427 	/* This is fun. The caller may want a bh back, or it may
2428 	 * not. ocfs2_inode_lock_update definitely wants one in, but
2429 	 * may or may not read one, depending on what's in the
2430 	 * LVB. The result of all of this is that we've *only* gone to
2431 	 * disk if we have to, so the complexity is worthwhile. */
2432 	status = ocfs2_inode_lock_update(inode, &local_bh);
2433 	if (status < 0) {
2434 		if (status != -ENOENT)
2435 			mlog_errno(status);
2436 		goto bail;
2437 	}
2438 getbh:
2439 	if (ret_bh) {
2440 		status = ocfs2_assign_bh(inode, ret_bh, local_bh);
2441 		if (status < 0) {
2442 			mlog_errno(status);
2443 			goto bail;
2444 		}
2445 	}
2446 
2447 bail:
2448 	if (status < 0) {
2449 		if (ret_bh && (*ret_bh)) {
2450 			brelse(*ret_bh);
2451 			*ret_bh = NULL;
2452 		}
2453 		if (acquired)
2454 			ocfs2_inode_unlock(inode, ex);
2455 	}
2456 
2457 	if (local_bh)
2458 		brelse(local_bh);
2459 
2460 	return status;
2461 }
2462 
2463 /*
2464  * This is working around a lock inversion between tasks acquiring DLM
2465  * locks while holding a page lock and the downconvert thread which
2466  * blocks dlm lock acquiry while acquiring page locks.
2467  *
2468  * ** These _with_page variantes are only intended to be called from aop
2469  * methods that hold page locks and return a very specific *positive* error
2470  * code that aop methods pass up to the VFS -- test for errors with != 0. **
2471  *
2472  * The DLM is called such that it returns -EAGAIN if it would have
2473  * blocked waiting for the downconvert thread.  In that case we unlock
2474  * our page so the downconvert thread can make progress.  Once we've
2475  * done this we have to return AOP_TRUNCATED_PAGE so the aop method
2476  * that called us can bubble that back up into the VFS who will then
2477  * immediately retry the aop call.
2478  */
2479 int ocfs2_inode_lock_with_page(struct inode *inode,
2480 			      struct buffer_head **ret_bh,
2481 			      int ex,
2482 			      struct page *page)
2483 {
2484 	int ret;
2485 
2486 	ret = ocfs2_inode_lock_full(inode, ret_bh, ex, OCFS2_LOCK_NONBLOCK);
2487 	if (ret == -EAGAIN) {
2488 		unlock_page(page);
2489 		ret = AOP_TRUNCATED_PAGE;
2490 	}
2491 
2492 	return ret;
2493 }
2494 
2495 int ocfs2_inode_lock_atime(struct inode *inode,
2496 			  struct vfsmount *vfsmnt,
2497 			  int *level)
2498 {
2499 	int ret;
2500 
2501 	ret = ocfs2_inode_lock(inode, NULL, 0);
2502 	if (ret < 0) {
2503 		mlog_errno(ret);
2504 		return ret;
2505 	}
2506 
2507 	/*
2508 	 * If we should update atime, we will get EX lock,
2509 	 * otherwise we just get PR lock.
2510 	 */
2511 	if (ocfs2_should_update_atime(inode, vfsmnt)) {
2512 		struct buffer_head *bh = NULL;
2513 
2514 		ocfs2_inode_unlock(inode, 0);
2515 		ret = ocfs2_inode_lock(inode, &bh, 1);
2516 		if (ret < 0) {
2517 			mlog_errno(ret);
2518 			return ret;
2519 		}
2520 		*level = 1;
2521 		if (ocfs2_should_update_atime(inode, vfsmnt))
2522 			ocfs2_update_inode_atime(inode, bh);
2523 		if (bh)
2524 			brelse(bh);
2525 	} else
2526 		*level = 0;
2527 
2528 	return ret;
2529 }
2530 
2531 void ocfs2_inode_unlock(struct inode *inode,
2532 		       int ex)
2533 {
2534 	int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2535 	struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_inode_lockres;
2536 	struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
2537 
2538 	mlog(0, "inode %llu drop %s META lock\n",
2539 	     (unsigned long long)OCFS2_I(inode)->ip_blkno,
2540 	     ex ? "EXMODE" : "PRMODE");
2541 
2542 	if (!ocfs2_is_hard_readonly(OCFS2_SB(inode->i_sb)) &&
2543 	    !ocfs2_mount_local(osb))
2544 		ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres, level);
2545 }
2546 
2547 /*
2548  * This _tracker variantes are introduced to deal with the recursive cluster
2549  * locking issue. The idea is to keep track of a lock holder on the stack of
2550  * the current process. If there's a lock holder on the stack, we know the
2551  * task context is already protected by cluster locking. Currently, they're
2552  * used in some VFS entry routines.
2553  *
2554  * return < 0 on error, return == 0 if there's no lock holder on the stack
2555  * before this call, return == 1 if this call would be a recursive locking.
2556  */
2557 int ocfs2_inode_lock_tracker(struct inode *inode,
2558 			     struct buffer_head **ret_bh,
2559 			     int ex,
2560 			     struct ocfs2_lock_holder *oh)
2561 {
2562 	int status;
2563 	int arg_flags = 0, has_locked;
2564 	struct ocfs2_lock_res *lockres;
2565 
2566 	lockres = &OCFS2_I(inode)->ip_inode_lockres;
2567 	has_locked = ocfs2_is_locked_by_me(lockres);
2568 	/* Just get buffer head if the cluster lock has been taken */
2569 	if (has_locked)
2570 		arg_flags = OCFS2_META_LOCK_GETBH;
2571 
2572 	if (likely(!has_locked || ret_bh)) {
2573 		status = ocfs2_inode_lock_full(inode, ret_bh, ex, arg_flags);
2574 		if (status < 0) {
2575 			if (status != -ENOENT)
2576 				mlog_errno(status);
2577 			return status;
2578 		}
2579 	}
2580 	if (!has_locked)
2581 		ocfs2_add_holder(lockres, oh);
2582 
2583 	return has_locked;
2584 }
2585 
2586 void ocfs2_inode_unlock_tracker(struct inode *inode,
2587 				int ex,
2588 				struct ocfs2_lock_holder *oh,
2589 				int had_lock)
2590 {
2591 	struct ocfs2_lock_res *lockres;
2592 
2593 	lockres = &OCFS2_I(inode)->ip_inode_lockres;
2594 	if (!had_lock) {
2595 		ocfs2_remove_holder(lockres, oh);
2596 		ocfs2_inode_unlock(inode, ex);
2597 	}
2598 }
2599 
2600 int ocfs2_orphan_scan_lock(struct ocfs2_super *osb, u32 *seqno)
2601 {
2602 	struct ocfs2_lock_res *lockres;
2603 	struct ocfs2_orphan_scan_lvb *lvb;
2604 	int status = 0;
2605 
2606 	if (ocfs2_is_hard_readonly(osb))
2607 		return -EROFS;
2608 
2609 	if (ocfs2_mount_local(osb))
2610 		return 0;
2611 
2612 	lockres = &osb->osb_orphan_scan.os_lockres;
2613 	status = ocfs2_cluster_lock(osb, lockres, DLM_LOCK_EX, 0, 0);
2614 	if (status < 0)
2615 		return status;
2616 
2617 	lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2618 	if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
2619 	    lvb->lvb_version == OCFS2_ORPHAN_LVB_VERSION)
2620 		*seqno = be32_to_cpu(lvb->lvb_os_seqno);
2621 	else
2622 		*seqno = osb->osb_orphan_scan.os_seqno + 1;
2623 
2624 	return status;
2625 }
2626 
2627 void ocfs2_orphan_scan_unlock(struct ocfs2_super *osb, u32 seqno)
2628 {
2629 	struct ocfs2_lock_res *lockres;
2630 	struct ocfs2_orphan_scan_lvb *lvb;
2631 
2632 	if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb)) {
2633 		lockres = &osb->osb_orphan_scan.os_lockres;
2634 		lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2635 		lvb->lvb_version = OCFS2_ORPHAN_LVB_VERSION;
2636 		lvb->lvb_os_seqno = cpu_to_be32(seqno);
2637 		ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
2638 	}
2639 }
2640 
2641 int ocfs2_super_lock(struct ocfs2_super *osb,
2642 		     int ex)
2643 {
2644 	int status = 0;
2645 	int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2646 	struct ocfs2_lock_res *lockres = &osb->osb_super_lockres;
2647 
2648 	if (ocfs2_is_hard_readonly(osb))
2649 		return -EROFS;
2650 
2651 	if (ocfs2_mount_local(osb))
2652 		goto bail;
2653 
2654 	status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
2655 	if (status < 0) {
2656 		mlog_errno(status);
2657 		goto bail;
2658 	}
2659 
2660 	/* The super block lock path is really in the best position to
2661 	 * know when resources covered by the lock need to be
2662 	 * refreshed, so we do it here. Of course, making sense of
2663 	 * everything is up to the caller :) */
2664 	status = ocfs2_should_refresh_lock_res(lockres);
2665 	if (status) {
2666 		status = ocfs2_refresh_slot_info(osb);
2667 
2668 		ocfs2_complete_lock_res_refresh(lockres, status);
2669 
2670 		if (status < 0) {
2671 			ocfs2_cluster_unlock(osb, lockres, level);
2672 			mlog_errno(status);
2673 		}
2674 		ocfs2_track_lock_refresh(lockres);
2675 	}
2676 bail:
2677 	return status;
2678 }
2679 
2680 void ocfs2_super_unlock(struct ocfs2_super *osb,
2681 			int ex)
2682 {
2683 	int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2684 	struct ocfs2_lock_res *lockres = &osb->osb_super_lockres;
2685 
2686 	if (!ocfs2_mount_local(osb))
2687 		ocfs2_cluster_unlock(osb, lockres, level);
2688 }
2689 
2690 int ocfs2_rename_lock(struct ocfs2_super *osb)
2691 {
2692 	int status;
2693 	struct ocfs2_lock_res *lockres = &osb->osb_rename_lockres;
2694 
2695 	if (ocfs2_is_hard_readonly(osb))
2696 		return -EROFS;
2697 
2698 	if (ocfs2_mount_local(osb))
2699 		return 0;
2700 
2701 	status = ocfs2_cluster_lock(osb, lockres, DLM_LOCK_EX, 0, 0);
2702 	if (status < 0)
2703 		mlog_errno(status);
2704 
2705 	return status;
2706 }
2707 
2708 void ocfs2_rename_unlock(struct ocfs2_super *osb)
2709 {
2710 	struct ocfs2_lock_res *lockres = &osb->osb_rename_lockres;
2711 
2712 	if (!ocfs2_mount_local(osb))
2713 		ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
2714 }
2715 
2716 int ocfs2_nfs_sync_lock(struct ocfs2_super *osb, int ex)
2717 {
2718 	int status;
2719 	struct ocfs2_lock_res *lockres = &osb->osb_nfs_sync_lockres;
2720 
2721 	if (ocfs2_is_hard_readonly(osb))
2722 		return -EROFS;
2723 
2724 	if (ocfs2_mount_local(osb))
2725 		return 0;
2726 
2727 	status = ocfs2_cluster_lock(osb, lockres, ex ? LKM_EXMODE : LKM_PRMODE,
2728 				    0, 0);
2729 	if (status < 0)
2730 		mlog(ML_ERROR, "lock on nfs sync lock failed %d\n", status);
2731 
2732 	return status;
2733 }
2734 
2735 void ocfs2_nfs_sync_unlock(struct ocfs2_super *osb, int ex)
2736 {
2737 	struct ocfs2_lock_res *lockres = &osb->osb_nfs_sync_lockres;
2738 
2739 	if (!ocfs2_mount_local(osb))
2740 		ocfs2_cluster_unlock(osb, lockres,
2741 				     ex ? LKM_EXMODE : LKM_PRMODE);
2742 }
2743 
2744 int ocfs2_dentry_lock(struct dentry *dentry, int ex)
2745 {
2746 	int ret;
2747 	int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2748 	struct ocfs2_dentry_lock *dl = dentry->d_fsdata;
2749 	struct ocfs2_super *osb = OCFS2_SB(dentry->d_sb);
2750 
2751 	BUG_ON(!dl);
2752 
2753 	if (ocfs2_is_hard_readonly(osb)) {
2754 		if (ex)
2755 			return -EROFS;
2756 		return 0;
2757 	}
2758 
2759 	if (ocfs2_mount_local(osb))
2760 		return 0;
2761 
2762 	ret = ocfs2_cluster_lock(osb, &dl->dl_lockres, level, 0, 0);
2763 	if (ret < 0)
2764 		mlog_errno(ret);
2765 
2766 	return ret;
2767 }
2768 
2769 void ocfs2_dentry_unlock(struct dentry *dentry, int ex)
2770 {
2771 	int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2772 	struct ocfs2_dentry_lock *dl = dentry->d_fsdata;
2773 	struct ocfs2_super *osb = OCFS2_SB(dentry->d_sb);
2774 
2775 	if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb))
2776 		ocfs2_cluster_unlock(osb, &dl->dl_lockres, level);
2777 }
2778 
2779 /* Reference counting of the dlm debug structure. We want this because
2780  * open references on the debug inodes can live on after a mount, so
2781  * we can't rely on the ocfs2_super to always exist. */
2782 static void ocfs2_dlm_debug_free(struct kref *kref)
2783 {
2784 	struct ocfs2_dlm_debug *dlm_debug;
2785 
2786 	dlm_debug = container_of(kref, struct ocfs2_dlm_debug, d_refcnt);
2787 
2788 	kfree(dlm_debug);
2789 }
2790 
2791 void ocfs2_put_dlm_debug(struct ocfs2_dlm_debug *dlm_debug)
2792 {
2793 	if (dlm_debug)
2794 		kref_put(&dlm_debug->d_refcnt, ocfs2_dlm_debug_free);
2795 }
2796 
2797 static void ocfs2_get_dlm_debug(struct ocfs2_dlm_debug *debug)
2798 {
2799 	kref_get(&debug->d_refcnt);
2800 }
2801 
2802 struct ocfs2_dlm_debug *ocfs2_new_dlm_debug(void)
2803 {
2804 	struct ocfs2_dlm_debug *dlm_debug;
2805 
2806 	dlm_debug = kmalloc(sizeof(struct ocfs2_dlm_debug), GFP_KERNEL);
2807 	if (!dlm_debug) {
2808 		mlog_errno(-ENOMEM);
2809 		goto out;
2810 	}
2811 
2812 	kref_init(&dlm_debug->d_refcnt);
2813 	INIT_LIST_HEAD(&dlm_debug->d_lockres_tracking);
2814 	dlm_debug->d_locking_state = NULL;
2815 out:
2816 	return dlm_debug;
2817 }
2818 
2819 /* Access to this is arbitrated for us via seq_file->sem. */
2820 struct ocfs2_dlm_seq_priv {
2821 	struct ocfs2_dlm_debug *p_dlm_debug;
2822 	struct ocfs2_lock_res p_iter_res;
2823 	struct ocfs2_lock_res p_tmp_res;
2824 };
2825 
2826 static struct ocfs2_lock_res *ocfs2_dlm_next_res(struct ocfs2_lock_res *start,
2827 						 struct ocfs2_dlm_seq_priv *priv)
2828 {
2829 	struct ocfs2_lock_res *iter, *ret = NULL;
2830 	struct ocfs2_dlm_debug *dlm_debug = priv->p_dlm_debug;
2831 
2832 	assert_spin_locked(&ocfs2_dlm_tracking_lock);
2833 
2834 	list_for_each_entry(iter, &start->l_debug_list, l_debug_list) {
2835 		/* discover the head of the list */
2836 		if (&iter->l_debug_list == &dlm_debug->d_lockres_tracking) {
2837 			mlog(0, "End of list found, %p\n", ret);
2838 			break;
2839 		}
2840 
2841 		/* We track our "dummy" iteration lockres' by a NULL
2842 		 * l_ops field. */
2843 		if (iter->l_ops != NULL) {
2844 			ret = iter;
2845 			break;
2846 		}
2847 	}
2848 
2849 	return ret;
2850 }
2851 
2852 static void *ocfs2_dlm_seq_start(struct seq_file *m, loff_t *pos)
2853 {
2854 	struct ocfs2_dlm_seq_priv *priv = m->private;
2855 	struct ocfs2_lock_res *iter;
2856 
2857 	spin_lock(&ocfs2_dlm_tracking_lock);
2858 	iter = ocfs2_dlm_next_res(&priv->p_iter_res, priv);
2859 	if (iter) {
2860 		/* Since lockres' have the lifetime of their container
2861 		 * (which can be inodes, ocfs2_supers, etc) we want to
2862 		 * copy this out to a temporary lockres while still
2863 		 * under the spinlock. Obviously after this we can't
2864 		 * trust any pointers on the copy returned, but that's
2865 		 * ok as the information we want isn't typically held
2866 		 * in them. */
2867 		priv->p_tmp_res = *iter;
2868 		iter = &priv->p_tmp_res;
2869 	}
2870 	spin_unlock(&ocfs2_dlm_tracking_lock);
2871 
2872 	return iter;
2873 }
2874 
2875 static void ocfs2_dlm_seq_stop(struct seq_file *m, void *v)
2876 {
2877 }
2878 
2879 static void *ocfs2_dlm_seq_next(struct seq_file *m, void *v, loff_t *pos)
2880 {
2881 	struct ocfs2_dlm_seq_priv *priv = m->private;
2882 	struct ocfs2_lock_res *iter = v;
2883 	struct ocfs2_lock_res *dummy = &priv->p_iter_res;
2884 
2885 	spin_lock(&ocfs2_dlm_tracking_lock);
2886 	iter = ocfs2_dlm_next_res(iter, priv);
2887 	list_del_init(&dummy->l_debug_list);
2888 	if (iter) {
2889 		list_add(&dummy->l_debug_list, &iter->l_debug_list);
2890 		priv->p_tmp_res = *iter;
2891 		iter = &priv->p_tmp_res;
2892 	}
2893 	spin_unlock(&ocfs2_dlm_tracking_lock);
2894 
2895 	return iter;
2896 }
2897 
2898 /*
2899  * Version is used by debugfs.ocfs2 to determine the format being used
2900  *
2901  * New in version 2
2902  *	- Lock stats printed
2903  * New in version 3
2904  *	- Max time in lock stats is in usecs (instead of nsecs)
2905  */
2906 #define OCFS2_DLM_DEBUG_STR_VERSION 3
2907 static int ocfs2_dlm_seq_show(struct seq_file *m, void *v)
2908 {
2909 	int i;
2910 	char *lvb;
2911 	struct ocfs2_lock_res *lockres = v;
2912 
2913 	if (!lockres)
2914 		return -EINVAL;
2915 
2916 	seq_printf(m, "0x%x\t", OCFS2_DLM_DEBUG_STR_VERSION);
2917 
2918 	if (lockres->l_type == OCFS2_LOCK_TYPE_DENTRY)
2919 		seq_printf(m, "%.*s%08x\t", OCFS2_DENTRY_LOCK_INO_START - 1,
2920 			   lockres->l_name,
2921 			   (unsigned int)ocfs2_get_dentry_lock_ino(lockres));
2922 	else
2923 		seq_printf(m, "%.*s\t", OCFS2_LOCK_ID_MAX_LEN, lockres->l_name);
2924 
2925 	seq_printf(m, "%d\t"
2926 		   "0x%lx\t"
2927 		   "0x%x\t"
2928 		   "0x%x\t"
2929 		   "%u\t"
2930 		   "%u\t"
2931 		   "%d\t"
2932 		   "%d\t",
2933 		   lockres->l_level,
2934 		   lockres->l_flags,
2935 		   lockres->l_action,
2936 		   lockres->l_unlock_action,
2937 		   lockres->l_ro_holders,
2938 		   lockres->l_ex_holders,
2939 		   lockres->l_requested,
2940 		   lockres->l_blocking);
2941 
2942 	/* Dump the raw LVB */
2943 	lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2944 	for(i = 0; i < DLM_LVB_LEN; i++)
2945 		seq_printf(m, "0x%x\t", lvb[i]);
2946 
2947 #ifdef CONFIG_OCFS2_FS_STATS
2948 # define lock_num_prmode(_l)		((_l)->l_lock_prmode.ls_gets)
2949 # define lock_num_exmode(_l)		((_l)->l_lock_exmode.ls_gets)
2950 # define lock_num_prmode_failed(_l)	((_l)->l_lock_prmode.ls_fail)
2951 # define lock_num_exmode_failed(_l)	((_l)->l_lock_exmode.ls_fail)
2952 # define lock_total_prmode(_l)		((_l)->l_lock_prmode.ls_total)
2953 # define lock_total_exmode(_l)		((_l)->l_lock_exmode.ls_total)
2954 # define lock_max_prmode(_l)		((_l)->l_lock_prmode.ls_max)
2955 # define lock_max_exmode(_l)		((_l)->l_lock_exmode.ls_max)
2956 # define lock_refresh(_l)		((_l)->l_lock_refresh)
2957 #else
2958 # define lock_num_prmode(_l)		(0)
2959 # define lock_num_exmode(_l)		(0)
2960 # define lock_num_prmode_failed(_l)	(0)
2961 # define lock_num_exmode_failed(_l)	(0)
2962 # define lock_total_prmode(_l)		(0ULL)
2963 # define lock_total_exmode(_l)		(0ULL)
2964 # define lock_max_prmode(_l)		(0)
2965 # define lock_max_exmode(_l)		(0)
2966 # define lock_refresh(_l)		(0)
2967 #endif
2968 	/* The following seq_print was added in version 2 of this output */
2969 	seq_printf(m, "%u\t"
2970 		   "%u\t"
2971 		   "%u\t"
2972 		   "%u\t"
2973 		   "%llu\t"
2974 		   "%llu\t"
2975 		   "%u\t"
2976 		   "%u\t"
2977 		   "%u\t",
2978 		   lock_num_prmode(lockres),
2979 		   lock_num_exmode(lockres),
2980 		   lock_num_prmode_failed(lockres),
2981 		   lock_num_exmode_failed(lockres),
2982 		   lock_total_prmode(lockres),
2983 		   lock_total_exmode(lockres),
2984 		   lock_max_prmode(lockres),
2985 		   lock_max_exmode(lockres),
2986 		   lock_refresh(lockres));
2987 
2988 	/* End the line */
2989 	seq_printf(m, "\n");
2990 	return 0;
2991 }
2992 
2993 static const struct seq_operations ocfs2_dlm_seq_ops = {
2994 	.start =	ocfs2_dlm_seq_start,
2995 	.stop =		ocfs2_dlm_seq_stop,
2996 	.next =		ocfs2_dlm_seq_next,
2997 	.show =		ocfs2_dlm_seq_show,
2998 };
2999 
3000 static int ocfs2_dlm_debug_release(struct inode *inode, struct file *file)
3001 {
3002 	struct seq_file *seq = file->private_data;
3003 	struct ocfs2_dlm_seq_priv *priv = seq->private;
3004 	struct ocfs2_lock_res *res = &priv->p_iter_res;
3005 
3006 	ocfs2_remove_lockres_tracking(res);
3007 	ocfs2_put_dlm_debug(priv->p_dlm_debug);
3008 	return seq_release_private(inode, file);
3009 }
3010 
3011 static int ocfs2_dlm_debug_open(struct inode *inode, struct file *file)
3012 {
3013 	struct ocfs2_dlm_seq_priv *priv;
3014 	struct ocfs2_super *osb;
3015 
3016 	priv = __seq_open_private(file, &ocfs2_dlm_seq_ops, sizeof(*priv));
3017 	if (!priv) {
3018 		mlog_errno(-ENOMEM);
3019 		return -ENOMEM;
3020 	}
3021 
3022 	osb = inode->i_private;
3023 	ocfs2_get_dlm_debug(osb->osb_dlm_debug);
3024 	priv->p_dlm_debug = osb->osb_dlm_debug;
3025 	INIT_LIST_HEAD(&priv->p_iter_res.l_debug_list);
3026 
3027 	ocfs2_add_lockres_tracking(&priv->p_iter_res,
3028 				   priv->p_dlm_debug);
3029 
3030 	return 0;
3031 }
3032 
3033 static const struct file_operations ocfs2_dlm_debug_fops = {
3034 	.open =		ocfs2_dlm_debug_open,
3035 	.release =	ocfs2_dlm_debug_release,
3036 	.read =		seq_read,
3037 	.llseek =	seq_lseek,
3038 };
3039 
3040 static int ocfs2_dlm_init_debug(struct ocfs2_super *osb)
3041 {
3042 	int ret = 0;
3043 	struct ocfs2_dlm_debug *dlm_debug = osb->osb_dlm_debug;
3044 
3045 	dlm_debug->d_locking_state = debugfs_create_file("locking_state",
3046 							 S_IFREG|S_IRUSR,
3047 							 osb->osb_debug_root,
3048 							 osb,
3049 							 &ocfs2_dlm_debug_fops);
3050 	if (!dlm_debug->d_locking_state) {
3051 		ret = -EINVAL;
3052 		mlog(ML_ERROR,
3053 		     "Unable to create locking state debugfs file.\n");
3054 		goto out;
3055 	}
3056 
3057 	ocfs2_get_dlm_debug(dlm_debug);
3058 out:
3059 	return ret;
3060 }
3061 
3062 static void ocfs2_dlm_shutdown_debug(struct ocfs2_super *osb)
3063 {
3064 	struct ocfs2_dlm_debug *dlm_debug = osb->osb_dlm_debug;
3065 
3066 	if (dlm_debug) {
3067 		debugfs_remove(dlm_debug->d_locking_state);
3068 		ocfs2_put_dlm_debug(dlm_debug);
3069 	}
3070 }
3071 
3072 int ocfs2_dlm_init(struct ocfs2_super *osb)
3073 {
3074 	int status = 0;
3075 	struct ocfs2_cluster_connection *conn = NULL;
3076 
3077 	if (ocfs2_mount_local(osb)) {
3078 		osb->node_num = 0;
3079 		goto local;
3080 	}
3081 
3082 	status = ocfs2_dlm_init_debug(osb);
3083 	if (status < 0) {
3084 		mlog_errno(status);
3085 		goto bail;
3086 	}
3087 
3088 	/* launch downconvert thread */
3089 	osb->dc_task = kthread_run(ocfs2_downconvert_thread, osb, "ocfs2dc-%s",
3090 			osb->uuid_str);
3091 	if (IS_ERR(osb->dc_task)) {
3092 		status = PTR_ERR(osb->dc_task);
3093 		osb->dc_task = NULL;
3094 		mlog_errno(status);
3095 		goto bail;
3096 	}
3097 
3098 	/* for now, uuid == domain */
3099 	status = ocfs2_cluster_connect(osb->osb_cluster_stack,
3100 				       osb->osb_cluster_name,
3101 				       strlen(osb->osb_cluster_name),
3102 				       osb->uuid_str,
3103 				       strlen(osb->uuid_str),
3104 				       &lproto, ocfs2_do_node_down, osb,
3105 				       &conn);
3106 	if (status) {
3107 		mlog_errno(status);
3108 		goto bail;
3109 	}
3110 
3111 	status = ocfs2_cluster_this_node(conn, &osb->node_num);
3112 	if (status < 0) {
3113 		mlog_errno(status);
3114 		mlog(ML_ERROR,
3115 		     "could not find this host's node number\n");
3116 		ocfs2_cluster_disconnect(conn, 0);
3117 		goto bail;
3118 	}
3119 
3120 local:
3121 	ocfs2_super_lock_res_init(&osb->osb_super_lockres, osb);
3122 	ocfs2_rename_lock_res_init(&osb->osb_rename_lockres, osb);
3123 	ocfs2_nfs_sync_lock_res_init(&osb->osb_nfs_sync_lockres, osb);
3124 	ocfs2_orphan_scan_lock_res_init(&osb->osb_orphan_scan.os_lockres, osb);
3125 
3126 	osb->cconn = conn;
3127 bail:
3128 	if (status < 0) {
3129 		ocfs2_dlm_shutdown_debug(osb);
3130 		if (osb->dc_task)
3131 			kthread_stop(osb->dc_task);
3132 	}
3133 
3134 	return status;
3135 }
3136 
3137 void ocfs2_dlm_shutdown(struct ocfs2_super *osb,
3138 			int hangup_pending)
3139 {
3140 	ocfs2_drop_osb_locks(osb);
3141 
3142 	/*
3143 	 * Now that we have dropped all locks and ocfs2_dismount_volume()
3144 	 * has disabled recovery, the DLM won't be talking to us.  It's
3145 	 * safe to tear things down before disconnecting the cluster.
3146 	 */
3147 
3148 	if (osb->dc_task) {
3149 		kthread_stop(osb->dc_task);
3150 		osb->dc_task = NULL;
3151 	}
3152 
3153 	ocfs2_lock_res_free(&osb->osb_super_lockres);
3154 	ocfs2_lock_res_free(&osb->osb_rename_lockres);
3155 	ocfs2_lock_res_free(&osb->osb_nfs_sync_lockres);
3156 	ocfs2_lock_res_free(&osb->osb_orphan_scan.os_lockres);
3157 
3158 	ocfs2_cluster_disconnect(osb->cconn, hangup_pending);
3159 	osb->cconn = NULL;
3160 
3161 	ocfs2_dlm_shutdown_debug(osb);
3162 }
3163 
3164 static int ocfs2_drop_lock(struct ocfs2_super *osb,
3165 			   struct ocfs2_lock_res *lockres)
3166 {
3167 	int ret;
3168 	unsigned long flags;
3169 	u32 lkm_flags = 0;
3170 
3171 	/* We didn't get anywhere near actually using this lockres. */
3172 	if (!(lockres->l_flags & OCFS2_LOCK_INITIALIZED))
3173 		goto out;
3174 
3175 	if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
3176 		lkm_flags |= DLM_LKF_VALBLK;
3177 
3178 	spin_lock_irqsave(&lockres->l_lock, flags);
3179 
3180 	mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_FREEING),
3181 			"lockres %s, flags 0x%lx\n",
3182 			lockres->l_name, lockres->l_flags);
3183 
3184 	while (lockres->l_flags & OCFS2_LOCK_BUSY) {
3185 		mlog(0, "waiting on busy lock \"%s\": flags = %lx, action = "
3186 		     "%u, unlock_action = %u\n",
3187 		     lockres->l_name, lockres->l_flags, lockres->l_action,
3188 		     lockres->l_unlock_action);
3189 
3190 		spin_unlock_irqrestore(&lockres->l_lock, flags);
3191 
3192 		/* XXX: Today we just wait on any busy
3193 		 * locks... Perhaps we need to cancel converts in the
3194 		 * future? */
3195 		ocfs2_wait_on_busy_lock(lockres);
3196 
3197 		spin_lock_irqsave(&lockres->l_lock, flags);
3198 	}
3199 
3200 	if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
3201 		if (lockres->l_flags & OCFS2_LOCK_ATTACHED &&
3202 		    lockres->l_level == DLM_LOCK_EX &&
3203 		    !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
3204 			lockres->l_ops->set_lvb(lockres);
3205 	}
3206 
3207 	if (lockres->l_flags & OCFS2_LOCK_BUSY)
3208 		mlog(ML_ERROR, "destroying busy lock: \"%s\"\n",
3209 		     lockres->l_name);
3210 	if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
3211 		mlog(0, "destroying blocked lock: \"%s\"\n", lockres->l_name);
3212 
3213 	if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
3214 		spin_unlock_irqrestore(&lockres->l_lock, flags);
3215 		goto out;
3216 	}
3217 
3218 	lockres_clear_flags(lockres, OCFS2_LOCK_ATTACHED);
3219 
3220 	/* make sure we never get here while waiting for an ast to
3221 	 * fire. */
3222 	BUG_ON(lockres->l_action != OCFS2_AST_INVALID);
3223 
3224 	/* is this necessary? */
3225 	lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
3226 	lockres->l_unlock_action = OCFS2_UNLOCK_DROP_LOCK;
3227 	spin_unlock_irqrestore(&lockres->l_lock, flags);
3228 
3229 	mlog(0, "lock %s\n", lockres->l_name);
3230 
3231 	ret = ocfs2_dlm_unlock(osb->cconn, &lockres->l_lksb, lkm_flags);
3232 	if (ret) {
3233 		ocfs2_log_dlm_error("ocfs2_dlm_unlock", ret, lockres);
3234 		mlog(ML_ERROR, "lockres flags: %lu\n", lockres->l_flags);
3235 		ocfs2_dlm_dump_lksb(&lockres->l_lksb);
3236 		BUG();
3237 	}
3238 	mlog(0, "lock %s, successful return from ocfs2_dlm_unlock\n",
3239 	     lockres->l_name);
3240 
3241 	ocfs2_wait_on_busy_lock(lockres);
3242 out:
3243 	return 0;
3244 }
3245 
3246 static void ocfs2_process_blocked_lock(struct ocfs2_super *osb,
3247 				       struct ocfs2_lock_res *lockres);
3248 
3249 /* Mark the lockres as being dropped. It will no longer be
3250  * queued if blocking, but we still may have to wait on it
3251  * being dequeued from the downconvert thread before we can consider
3252  * it safe to drop.
3253  *
3254  * You can *not* attempt to call cluster_lock on this lockres anymore. */
3255 void ocfs2_mark_lockres_freeing(struct ocfs2_super *osb,
3256 				struct ocfs2_lock_res *lockres)
3257 {
3258 	int status;
3259 	struct ocfs2_mask_waiter mw;
3260 	unsigned long flags, flags2;
3261 
3262 	ocfs2_init_mask_waiter(&mw);
3263 
3264 	spin_lock_irqsave(&lockres->l_lock, flags);
3265 	lockres->l_flags |= OCFS2_LOCK_FREEING;
3266 	if (lockres->l_flags & OCFS2_LOCK_QUEUED && current == osb->dc_task) {
3267 		/*
3268 		 * We know the downconvert is queued but not in progress
3269 		 * because we are the downconvert thread and processing
3270 		 * different lock. So we can just remove the lock from the
3271 		 * queue. This is not only an optimization but also a way
3272 		 * to avoid the following deadlock:
3273 		 *   ocfs2_dentry_post_unlock()
3274 		 *     ocfs2_dentry_lock_put()
3275 		 *       ocfs2_drop_dentry_lock()
3276 		 *         iput()
3277 		 *           ocfs2_evict_inode()
3278 		 *             ocfs2_clear_inode()
3279 		 *               ocfs2_mark_lockres_freeing()
3280 		 *                 ... blocks waiting for OCFS2_LOCK_QUEUED
3281 		 *                 since we are the downconvert thread which
3282 		 *                 should clear the flag.
3283 		 */
3284 		spin_unlock_irqrestore(&lockres->l_lock, flags);
3285 		spin_lock_irqsave(&osb->dc_task_lock, flags2);
3286 		list_del_init(&lockres->l_blocked_list);
3287 		osb->blocked_lock_count--;
3288 		spin_unlock_irqrestore(&osb->dc_task_lock, flags2);
3289 		/*
3290 		 * Warn if we recurse into another post_unlock call.  Strictly
3291 		 * speaking it isn't a problem but we need to be careful if
3292 		 * that happens (stack overflow, deadlocks, ...) so warn if
3293 		 * ocfs2 grows a path for which this can happen.
3294 		 */
3295 		WARN_ON_ONCE(lockres->l_ops->post_unlock);
3296 		/* Since the lock is freeing we don't do much in the fn below */
3297 		ocfs2_process_blocked_lock(osb, lockres);
3298 		return;
3299 	}
3300 	while (lockres->l_flags & OCFS2_LOCK_QUEUED) {
3301 		lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_QUEUED, 0);
3302 		spin_unlock_irqrestore(&lockres->l_lock, flags);
3303 
3304 		mlog(0, "Waiting on lockres %s\n", lockres->l_name);
3305 
3306 		status = ocfs2_wait_for_mask(&mw);
3307 		if (status)
3308 			mlog_errno(status);
3309 
3310 		spin_lock_irqsave(&lockres->l_lock, flags);
3311 	}
3312 	spin_unlock_irqrestore(&lockres->l_lock, flags);
3313 }
3314 
3315 void ocfs2_simple_drop_lockres(struct ocfs2_super *osb,
3316 			       struct ocfs2_lock_res *lockres)
3317 {
3318 	int ret;
3319 
3320 	ocfs2_mark_lockres_freeing(osb, lockres);
3321 	ret = ocfs2_drop_lock(osb, lockres);
3322 	if (ret)
3323 		mlog_errno(ret);
3324 }
3325 
3326 static void ocfs2_drop_osb_locks(struct ocfs2_super *osb)
3327 {
3328 	ocfs2_simple_drop_lockres(osb, &osb->osb_super_lockres);
3329 	ocfs2_simple_drop_lockres(osb, &osb->osb_rename_lockres);
3330 	ocfs2_simple_drop_lockres(osb, &osb->osb_nfs_sync_lockres);
3331 	ocfs2_simple_drop_lockres(osb, &osb->osb_orphan_scan.os_lockres);
3332 }
3333 
3334 int ocfs2_drop_inode_locks(struct inode *inode)
3335 {
3336 	int status, err;
3337 
3338 	/* No need to call ocfs2_mark_lockres_freeing here -
3339 	 * ocfs2_clear_inode has done it for us. */
3340 
3341 	err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3342 			      &OCFS2_I(inode)->ip_open_lockres);
3343 	if (err < 0)
3344 		mlog_errno(err);
3345 
3346 	status = err;
3347 
3348 	err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3349 			      &OCFS2_I(inode)->ip_inode_lockres);
3350 	if (err < 0)
3351 		mlog_errno(err);
3352 	if (err < 0 && !status)
3353 		status = err;
3354 
3355 	err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3356 			      &OCFS2_I(inode)->ip_rw_lockres);
3357 	if (err < 0)
3358 		mlog_errno(err);
3359 	if (err < 0 && !status)
3360 		status = err;
3361 
3362 	return status;
3363 }
3364 
3365 static unsigned int ocfs2_prepare_downconvert(struct ocfs2_lock_res *lockres,
3366 					      int new_level)
3367 {
3368 	assert_spin_locked(&lockres->l_lock);
3369 
3370 	BUG_ON(lockres->l_blocking <= DLM_LOCK_NL);
3371 
3372 	if (lockres->l_level <= new_level) {
3373 		mlog(ML_ERROR, "lockres %s, lvl %d <= %d, blcklst %d, mask %d, "
3374 		     "type %d, flags 0x%lx, hold %d %d, act %d %d, req %d, "
3375 		     "block %d, pgen %d\n", lockres->l_name, lockres->l_level,
3376 		     new_level, list_empty(&lockres->l_blocked_list),
3377 		     list_empty(&lockres->l_mask_waiters), lockres->l_type,
3378 		     lockres->l_flags, lockres->l_ro_holders,
3379 		     lockres->l_ex_holders, lockres->l_action,
3380 		     lockres->l_unlock_action, lockres->l_requested,
3381 		     lockres->l_blocking, lockres->l_pending_gen);
3382 		BUG();
3383 	}
3384 
3385 	mlog(ML_BASTS, "lockres %s, level %d => %d, blocking %d\n",
3386 	     lockres->l_name, lockres->l_level, new_level, lockres->l_blocking);
3387 
3388 	lockres->l_action = OCFS2_AST_DOWNCONVERT;
3389 	lockres->l_requested = new_level;
3390 	lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
3391 	return lockres_set_pending(lockres);
3392 }
3393 
3394 static int ocfs2_downconvert_lock(struct ocfs2_super *osb,
3395 				  struct ocfs2_lock_res *lockres,
3396 				  int new_level,
3397 				  int lvb,
3398 				  unsigned int generation)
3399 {
3400 	int ret;
3401 	u32 dlm_flags = DLM_LKF_CONVERT;
3402 
3403 	mlog(ML_BASTS, "lockres %s, level %d => %d\n", lockres->l_name,
3404 	     lockres->l_level, new_level);
3405 
3406 	/*
3407 	 * On DLM_LKF_VALBLK, fsdlm behaves differently with o2cb. It always
3408 	 * expects DLM_LKF_VALBLK being set if the LKB has LVB, so that
3409 	 * we can recover correctly from node failure. Otherwise, we may get
3410 	 * invalid LVB in LKB, but without DLM_SBF_VALNOTVALID being set.
3411 	 */
3412 	if (!ocfs2_is_o2cb_active() &&
3413 	    lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
3414 		lvb = 1;
3415 
3416 	if (lvb)
3417 		dlm_flags |= DLM_LKF_VALBLK;
3418 
3419 	ret = ocfs2_dlm_lock(osb->cconn,
3420 			     new_level,
3421 			     &lockres->l_lksb,
3422 			     dlm_flags,
3423 			     lockres->l_name,
3424 			     OCFS2_LOCK_ID_MAX_LEN - 1);
3425 	lockres_clear_pending(lockres, generation, osb);
3426 	if (ret) {
3427 		ocfs2_log_dlm_error("ocfs2_dlm_lock", ret, lockres);
3428 		ocfs2_recover_from_dlm_error(lockres, 1);
3429 		goto bail;
3430 	}
3431 
3432 	ret = 0;
3433 bail:
3434 	return ret;
3435 }
3436 
3437 /* returns 1 when the caller should unlock and call ocfs2_dlm_unlock */
3438 static int ocfs2_prepare_cancel_convert(struct ocfs2_super *osb,
3439 				        struct ocfs2_lock_res *lockres)
3440 {
3441 	assert_spin_locked(&lockres->l_lock);
3442 
3443 	if (lockres->l_unlock_action == OCFS2_UNLOCK_CANCEL_CONVERT) {
3444 		/* If we're already trying to cancel a lock conversion
3445 		 * then just drop the spinlock and allow the caller to
3446 		 * requeue this lock. */
3447 		mlog(ML_BASTS, "lockres %s, skip convert\n", lockres->l_name);
3448 		return 0;
3449 	}
3450 
3451 	/* were we in a convert when we got the bast fire? */
3452 	BUG_ON(lockres->l_action != OCFS2_AST_CONVERT &&
3453 	       lockres->l_action != OCFS2_AST_DOWNCONVERT);
3454 	/* set things up for the unlockast to know to just
3455 	 * clear out the ast_action and unset busy, etc. */
3456 	lockres->l_unlock_action = OCFS2_UNLOCK_CANCEL_CONVERT;
3457 
3458 	mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_BUSY),
3459 			"lock %s, invalid flags: 0x%lx\n",
3460 			lockres->l_name, lockres->l_flags);
3461 
3462 	mlog(ML_BASTS, "lockres %s\n", lockres->l_name);
3463 
3464 	return 1;
3465 }
3466 
3467 static int ocfs2_cancel_convert(struct ocfs2_super *osb,
3468 				struct ocfs2_lock_res *lockres)
3469 {
3470 	int ret;
3471 
3472 	ret = ocfs2_dlm_unlock(osb->cconn, &lockres->l_lksb,
3473 			       DLM_LKF_CANCEL);
3474 	if (ret) {
3475 		ocfs2_log_dlm_error("ocfs2_dlm_unlock", ret, lockres);
3476 		ocfs2_recover_from_dlm_error(lockres, 0);
3477 	}
3478 
3479 	mlog(ML_BASTS, "lockres %s\n", lockres->l_name);
3480 
3481 	return ret;
3482 }
3483 
3484 static int ocfs2_unblock_lock(struct ocfs2_super *osb,
3485 			      struct ocfs2_lock_res *lockres,
3486 			      struct ocfs2_unblock_ctl *ctl)
3487 {
3488 	unsigned long flags;
3489 	int blocking;
3490 	int new_level;
3491 	int level;
3492 	int ret = 0;
3493 	int set_lvb = 0;
3494 	unsigned int gen;
3495 
3496 	spin_lock_irqsave(&lockres->l_lock, flags);
3497 
3498 recheck:
3499 	/*
3500 	 * Is it still blocking? If not, we have no more work to do.
3501 	 */
3502 	if (!(lockres->l_flags & OCFS2_LOCK_BLOCKED)) {
3503 		BUG_ON(lockres->l_blocking != DLM_LOCK_NL);
3504 		spin_unlock_irqrestore(&lockres->l_lock, flags);
3505 		ret = 0;
3506 		goto leave;
3507 	}
3508 
3509 	if (lockres->l_flags & OCFS2_LOCK_BUSY) {
3510 		/* XXX
3511 		 * This is a *big* race.  The OCFS2_LOCK_PENDING flag
3512 		 * exists entirely for one reason - another thread has set
3513 		 * OCFS2_LOCK_BUSY, but has *NOT* yet called dlm_lock().
3514 		 *
3515 		 * If we do ocfs2_cancel_convert() before the other thread
3516 		 * calls dlm_lock(), our cancel will do nothing.  We will
3517 		 * get no ast, and we will have no way of knowing the
3518 		 * cancel failed.  Meanwhile, the other thread will call
3519 		 * into dlm_lock() and wait...forever.
3520 		 *
3521 		 * Why forever?  Because another node has asked for the
3522 		 * lock first; that's why we're here in unblock_lock().
3523 		 *
3524 		 * The solution is OCFS2_LOCK_PENDING.  When PENDING is
3525 		 * set, we just requeue the unblock.  Only when the other
3526 		 * thread has called dlm_lock() and cleared PENDING will
3527 		 * we then cancel their request.
3528 		 *
3529 		 * All callers of dlm_lock() must set OCFS2_DLM_PENDING
3530 		 * at the same time they set OCFS2_DLM_BUSY.  They must
3531 		 * clear OCFS2_DLM_PENDING after dlm_lock() returns.
3532 		 */
3533 		if (lockres->l_flags & OCFS2_LOCK_PENDING) {
3534 			mlog(ML_BASTS, "lockres %s, ReQ: Pending\n",
3535 			     lockres->l_name);
3536 			goto leave_requeue;
3537 		}
3538 
3539 		ctl->requeue = 1;
3540 		ret = ocfs2_prepare_cancel_convert(osb, lockres);
3541 		spin_unlock_irqrestore(&lockres->l_lock, flags);
3542 		if (ret) {
3543 			ret = ocfs2_cancel_convert(osb, lockres);
3544 			if (ret < 0)
3545 				mlog_errno(ret);
3546 		}
3547 		goto leave;
3548 	}
3549 
3550 	/*
3551 	 * This prevents livelocks. OCFS2_LOCK_UPCONVERT_FINISHING flag is
3552 	 * set when the ast is received for an upconvert just before the
3553 	 * OCFS2_LOCK_BUSY flag is cleared. Now if the fs received a bast
3554 	 * on the heels of the ast, we want to delay the downconvert just
3555 	 * enough to allow the up requestor to do its task. Because this
3556 	 * lock is in the blocked queue, the lock will be downconverted
3557 	 * as soon as the requestor is done with the lock.
3558 	 */
3559 	if (lockres->l_flags & OCFS2_LOCK_UPCONVERT_FINISHING)
3560 		goto leave_requeue;
3561 
3562 	/*
3563 	 * How can we block and yet be at NL?  We were trying to upconvert
3564 	 * from NL and got canceled.  The code comes back here, and now
3565 	 * we notice and clear BLOCKING.
3566 	 */
3567 	if (lockres->l_level == DLM_LOCK_NL) {
3568 		BUG_ON(lockres->l_ex_holders || lockres->l_ro_holders);
3569 		mlog(ML_BASTS, "lockres %s, Aborting dc\n", lockres->l_name);
3570 		lockres->l_blocking = DLM_LOCK_NL;
3571 		lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED);
3572 		spin_unlock_irqrestore(&lockres->l_lock, flags);
3573 		goto leave;
3574 	}
3575 
3576 	/* if we're blocking an exclusive and we have *any* holders,
3577 	 * then requeue. */
3578 	if ((lockres->l_blocking == DLM_LOCK_EX)
3579 	    && (lockres->l_ex_holders || lockres->l_ro_holders)) {
3580 		mlog(ML_BASTS, "lockres %s, ReQ: EX/PR Holders %u,%u\n",
3581 		     lockres->l_name, lockres->l_ex_holders,
3582 		     lockres->l_ro_holders);
3583 		goto leave_requeue;
3584 	}
3585 
3586 	/* If it's a PR we're blocking, then only
3587 	 * requeue if we've got any EX holders */
3588 	if (lockres->l_blocking == DLM_LOCK_PR &&
3589 	    lockres->l_ex_holders) {
3590 		mlog(ML_BASTS, "lockres %s, ReQ: EX Holders %u\n",
3591 		     lockres->l_name, lockres->l_ex_holders);
3592 		goto leave_requeue;
3593 	}
3594 
3595 	/*
3596 	 * Can we get a lock in this state if the holder counts are
3597 	 * zero? The meta data unblock code used to check this.
3598 	 */
3599 	if ((lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
3600 	    && (lockres->l_flags & OCFS2_LOCK_REFRESHING)) {
3601 		mlog(ML_BASTS, "lockres %s, ReQ: Lock Refreshing\n",
3602 		     lockres->l_name);
3603 		goto leave_requeue;
3604 	}
3605 
3606 	new_level = ocfs2_highest_compat_lock_level(lockres->l_blocking);
3607 
3608 	if (lockres->l_ops->check_downconvert
3609 	    && !lockres->l_ops->check_downconvert(lockres, new_level)) {
3610 		mlog(ML_BASTS, "lockres %s, ReQ: Checkpointing\n",
3611 		     lockres->l_name);
3612 		goto leave_requeue;
3613 	}
3614 
3615 	/* If we get here, then we know that there are no more
3616 	 * incompatible holders (and anyone asking for an incompatible
3617 	 * lock is blocked). We can now downconvert the lock */
3618 	if (!lockres->l_ops->downconvert_worker)
3619 		goto downconvert;
3620 
3621 	/* Some lockres types want to do a bit of work before
3622 	 * downconverting a lock. Allow that here. The worker function
3623 	 * may sleep, so we save off a copy of what we're blocking as
3624 	 * it may change while we're not holding the spin lock. */
3625 	blocking = lockres->l_blocking;
3626 	level = lockres->l_level;
3627 	spin_unlock_irqrestore(&lockres->l_lock, flags);
3628 
3629 	ctl->unblock_action = lockres->l_ops->downconvert_worker(lockres, blocking);
3630 
3631 	if (ctl->unblock_action == UNBLOCK_STOP_POST) {
3632 		mlog(ML_BASTS, "lockres %s, UNBLOCK_STOP_POST\n",
3633 		     lockres->l_name);
3634 		goto leave;
3635 	}
3636 
3637 	spin_lock_irqsave(&lockres->l_lock, flags);
3638 	if ((blocking != lockres->l_blocking) || (level != lockres->l_level)) {
3639 		/* If this changed underneath us, then we can't drop
3640 		 * it just yet. */
3641 		mlog(ML_BASTS, "lockres %s, block=%d:%d, level=%d:%d, "
3642 		     "Recheck\n", lockres->l_name, blocking,
3643 		     lockres->l_blocking, level, lockres->l_level);
3644 		goto recheck;
3645 	}
3646 
3647 downconvert:
3648 	ctl->requeue = 0;
3649 
3650 	if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
3651 		if (lockres->l_level == DLM_LOCK_EX)
3652 			set_lvb = 1;
3653 
3654 		/*
3655 		 * We only set the lvb if the lock has been fully
3656 		 * refreshed - otherwise we risk setting stale
3657 		 * data. Otherwise, there's no need to actually clear
3658 		 * out the lvb here as it's value is still valid.
3659 		 */
3660 		if (set_lvb && !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
3661 			lockres->l_ops->set_lvb(lockres);
3662 	}
3663 
3664 	gen = ocfs2_prepare_downconvert(lockres, new_level);
3665 	spin_unlock_irqrestore(&lockres->l_lock, flags);
3666 	ret = ocfs2_downconvert_lock(osb, lockres, new_level, set_lvb,
3667 				     gen);
3668 
3669 leave:
3670 	if (ret)
3671 		mlog_errno(ret);
3672 	return ret;
3673 
3674 leave_requeue:
3675 	spin_unlock_irqrestore(&lockres->l_lock, flags);
3676 	ctl->requeue = 1;
3677 
3678 	return 0;
3679 }
3680 
3681 static int ocfs2_data_convert_worker(struct ocfs2_lock_res *lockres,
3682 				     int blocking)
3683 {
3684 	struct inode *inode;
3685 	struct address_space *mapping;
3686 	struct ocfs2_inode_info *oi;
3687 
3688        	inode = ocfs2_lock_res_inode(lockres);
3689 	mapping = inode->i_mapping;
3690 
3691 	if (S_ISDIR(inode->i_mode)) {
3692 		oi = OCFS2_I(inode);
3693 		oi->ip_dir_lock_gen++;
3694 		mlog(0, "generation: %u\n", oi->ip_dir_lock_gen);
3695 		goto out;
3696 	}
3697 
3698 	if (!S_ISREG(inode->i_mode))
3699 		goto out;
3700 
3701 	/*
3702 	 * We need this before the filemap_fdatawrite() so that it can
3703 	 * transfer the dirty bit from the PTE to the
3704 	 * page. Unfortunately this means that even for EX->PR
3705 	 * downconverts, we'll lose our mappings and have to build
3706 	 * them up again.
3707 	 */
3708 	unmap_mapping_range(mapping, 0, 0, 0);
3709 
3710 	if (filemap_fdatawrite(mapping)) {
3711 		mlog(ML_ERROR, "Could not sync inode %llu for downconvert!",
3712 		     (unsigned long long)OCFS2_I(inode)->ip_blkno);
3713 	}
3714 	sync_mapping_buffers(mapping);
3715 	if (blocking == DLM_LOCK_EX) {
3716 		truncate_inode_pages(mapping, 0);
3717 	} else {
3718 		/* We only need to wait on the I/O if we're not also
3719 		 * truncating pages because truncate_inode_pages waits
3720 		 * for us above. We don't truncate pages if we're
3721 		 * blocking anything < EXMODE because we want to keep
3722 		 * them around in that case. */
3723 		filemap_fdatawait(mapping);
3724 	}
3725 
3726 	forget_all_cached_acls(inode);
3727 
3728 out:
3729 	return UNBLOCK_CONTINUE;
3730 }
3731 
3732 static int ocfs2_ci_checkpointed(struct ocfs2_caching_info *ci,
3733 				 struct ocfs2_lock_res *lockres,
3734 				 int new_level)
3735 {
3736 	int checkpointed = ocfs2_ci_fully_checkpointed(ci);
3737 
3738 	BUG_ON(new_level != DLM_LOCK_NL && new_level != DLM_LOCK_PR);
3739 	BUG_ON(lockres->l_level != DLM_LOCK_EX && !checkpointed);
3740 
3741 	if (checkpointed)
3742 		return 1;
3743 
3744 	ocfs2_start_checkpoint(OCFS2_SB(ocfs2_metadata_cache_get_super(ci)));
3745 	return 0;
3746 }
3747 
3748 static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
3749 					int new_level)
3750 {
3751 	struct inode *inode = ocfs2_lock_res_inode(lockres);
3752 
3753 	return ocfs2_ci_checkpointed(INODE_CACHE(inode), lockres, new_level);
3754 }
3755 
3756 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres)
3757 {
3758 	struct inode *inode = ocfs2_lock_res_inode(lockres);
3759 
3760 	__ocfs2_stuff_meta_lvb(inode);
3761 }
3762 
3763 /*
3764  * Does the final reference drop on our dentry lock. Right now this
3765  * happens in the downconvert thread, but we could choose to simplify the
3766  * dlmglue API and push these off to the ocfs2_wq in the future.
3767  */
3768 static void ocfs2_dentry_post_unlock(struct ocfs2_super *osb,
3769 				     struct ocfs2_lock_res *lockres)
3770 {
3771 	struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
3772 	ocfs2_dentry_lock_put(osb, dl);
3773 }
3774 
3775 /*
3776  * d_delete() matching dentries before the lock downconvert.
3777  *
3778  * At this point, any process waiting to destroy the
3779  * dentry_lock due to last ref count is stopped by the
3780  * OCFS2_LOCK_QUEUED flag.
3781  *
3782  * We have two potential problems
3783  *
3784  * 1) If we do the last reference drop on our dentry_lock (via dput)
3785  *    we'll wind up in ocfs2_release_dentry_lock(), waiting on
3786  *    the downconvert to finish. Instead we take an elevated
3787  *    reference and push the drop until after we've completed our
3788  *    unblock processing.
3789  *
3790  * 2) There might be another process with a final reference,
3791  *    waiting on us to finish processing. If this is the case, we
3792  *    detect it and exit out - there's no more dentries anyway.
3793  */
3794 static int ocfs2_dentry_convert_worker(struct ocfs2_lock_res *lockres,
3795 				       int blocking)
3796 {
3797 	struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
3798 	struct ocfs2_inode_info *oi = OCFS2_I(dl->dl_inode);
3799 	struct dentry *dentry;
3800 	unsigned long flags;
3801 	int extra_ref = 0;
3802 
3803 	/*
3804 	 * This node is blocking another node from getting a read
3805 	 * lock. This happens when we've renamed within a
3806 	 * directory. We've forced the other nodes to d_delete(), but
3807 	 * we never actually dropped our lock because it's still
3808 	 * valid. The downconvert code will retain a PR for this node,
3809 	 * so there's no further work to do.
3810 	 */
3811 	if (blocking == DLM_LOCK_PR)
3812 		return UNBLOCK_CONTINUE;
3813 
3814 	/*
3815 	 * Mark this inode as potentially orphaned. The code in
3816 	 * ocfs2_delete_inode() will figure out whether it actually
3817 	 * needs to be freed or not.
3818 	 */
3819 	spin_lock(&oi->ip_lock);
3820 	oi->ip_flags |= OCFS2_INODE_MAYBE_ORPHANED;
3821 	spin_unlock(&oi->ip_lock);
3822 
3823 	/*
3824 	 * Yuck. We need to make sure however that the check of
3825 	 * OCFS2_LOCK_FREEING and the extra reference are atomic with
3826 	 * respect to a reference decrement or the setting of that
3827 	 * flag.
3828 	 */
3829 	spin_lock_irqsave(&lockres->l_lock, flags);
3830 	spin_lock(&dentry_attach_lock);
3831 	if (!(lockres->l_flags & OCFS2_LOCK_FREEING)
3832 	    && dl->dl_count) {
3833 		dl->dl_count++;
3834 		extra_ref = 1;
3835 	}
3836 	spin_unlock(&dentry_attach_lock);
3837 	spin_unlock_irqrestore(&lockres->l_lock, flags);
3838 
3839 	mlog(0, "extra_ref = %d\n", extra_ref);
3840 
3841 	/*
3842 	 * We have a process waiting on us in ocfs2_dentry_iput(),
3843 	 * which means we can't have any more outstanding
3844 	 * aliases. There's no need to do any more work.
3845 	 */
3846 	if (!extra_ref)
3847 		return UNBLOCK_CONTINUE;
3848 
3849 	spin_lock(&dentry_attach_lock);
3850 	while (1) {
3851 		dentry = ocfs2_find_local_alias(dl->dl_inode,
3852 						dl->dl_parent_blkno, 1);
3853 		if (!dentry)
3854 			break;
3855 		spin_unlock(&dentry_attach_lock);
3856 
3857 		if (S_ISDIR(dl->dl_inode->i_mode))
3858 			shrink_dcache_parent(dentry);
3859 
3860 		mlog(0, "d_delete(%pd);\n", dentry);
3861 
3862 		/*
3863 		 * The following dcache calls may do an
3864 		 * iput(). Normally we don't want that from the
3865 		 * downconverting thread, but in this case it's ok
3866 		 * because the requesting node already has an
3867 		 * exclusive lock on the inode, so it can't be queued
3868 		 * for a downconvert.
3869 		 */
3870 		d_delete(dentry);
3871 		dput(dentry);
3872 
3873 		spin_lock(&dentry_attach_lock);
3874 	}
3875 	spin_unlock(&dentry_attach_lock);
3876 
3877 	/*
3878 	 * If we are the last holder of this dentry lock, there is no
3879 	 * reason to downconvert so skip straight to the unlock.
3880 	 */
3881 	if (dl->dl_count == 1)
3882 		return UNBLOCK_STOP_POST;
3883 
3884 	return UNBLOCK_CONTINUE_POST;
3885 }
3886 
3887 static int ocfs2_check_refcount_downconvert(struct ocfs2_lock_res *lockres,
3888 					    int new_level)
3889 {
3890 	struct ocfs2_refcount_tree *tree =
3891 				ocfs2_lock_res_refcount_tree(lockres);
3892 
3893 	return ocfs2_ci_checkpointed(&tree->rf_ci, lockres, new_level);
3894 }
3895 
3896 static int ocfs2_refcount_convert_worker(struct ocfs2_lock_res *lockres,
3897 					 int blocking)
3898 {
3899 	struct ocfs2_refcount_tree *tree =
3900 				ocfs2_lock_res_refcount_tree(lockres);
3901 
3902 	ocfs2_metadata_cache_purge(&tree->rf_ci);
3903 
3904 	return UNBLOCK_CONTINUE;
3905 }
3906 
3907 static void ocfs2_set_qinfo_lvb(struct ocfs2_lock_res *lockres)
3908 {
3909 	struct ocfs2_qinfo_lvb *lvb;
3910 	struct ocfs2_mem_dqinfo *oinfo = ocfs2_lock_res_qinfo(lockres);
3911 	struct mem_dqinfo *info = sb_dqinfo(oinfo->dqi_gi.dqi_sb,
3912 					    oinfo->dqi_gi.dqi_type);
3913 
3914 	lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
3915 	lvb->lvb_version = OCFS2_QINFO_LVB_VERSION;
3916 	lvb->lvb_bgrace = cpu_to_be32(info->dqi_bgrace);
3917 	lvb->lvb_igrace = cpu_to_be32(info->dqi_igrace);
3918 	lvb->lvb_syncms = cpu_to_be32(oinfo->dqi_syncms);
3919 	lvb->lvb_blocks = cpu_to_be32(oinfo->dqi_gi.dqi_blocks);
3920 	lvb->lvb_free_blk = cpu_to_be32(oinfo->dqi_gi.dqi_free_blk);
3921 	lvb->lvb_free_entry = cpu_to_be32(oinfo->dqi_gi.dqi_free_entry);
3922 }
3923 
3924 void ocfs2_qinfo_unlock(struct ocfs2_mem_dqinfo *oinfo, int ex)
3925 {
3926 	struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
3927 	struct ocfs2_super *osb = OCFS2_SB(oinfo->dqi_gi.dqi_sb);
3928 	int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
3929 
3930 	if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb))
3931 		ocfs2_cluster_unlock(osb, lockres, level);
3932 }
3933 
3934 static int ocfs2_refresh_qinfo(struct ocfs2_mem_dqinfo *oinfo)
3935 {
3936 	struct mem_dqinfo *info = sb_dqinfo(oinfo->dqi_gi.dqi_sb,
3937 					    oinfo->dqi_gi.dqi_type);
3938 	struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
3939 	struct ocfs2_qinfo_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
3940 	struct buffer_head *bh = NULL;
3941 	struct ocfs2_global_disk_dqinfo *gdinfo;
3942 	int status = 0;
3943 
3944 	if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
3945 	    lvb->lvb_version == OCFS2_QINFO_LVB_VERSION) {
3946 		info->dqi_bgrace = be32_to_cpu(lvb->lvb_bgrace);
3947 		info->dqi_igrace = be32_to_cpu(lvb->lvb_igrace);
3948 		oinfo->dqi_syncms = be32_to_cpu(lvb->lvb_syncms);
3949 		oinfo->dqi_gi.dqi_blocks = be32_to_cpu(lvb->lvb_blocks);
3950 		oinfo->dqi_gi.dqi_free_blk = be32_to_cpu(lvb->lvb_free_blk);
3951 		oinfo->dqi_gi.dqi_free_entry =
3952 					be32_to_cpu(lvb->lvb_free_entry);
3953 	} else {
3954 		status = ocfs2_read_quota_phys_block(oinfo->dqi_gqinode,
3955 						     oinfo->dqi_giblk, &bh);
3956 		if (status) {
3957 			mlog_errno(status);
3958 			goto bail;
3959 		}
3960 		gdinfo = (struct ocfs2_global_disk_dqinfo *)
3961 					(bh->b_data + OCFS2_GLOBAL_INFO_OFF);
3962 		info->dqi_bgrace = le32_to_cpu(gdinfo->dqi_bgrace);
3963 		info->dqi_igrace = le32_to_cpu(gdinfo->dqi_igrace);
3964 		oinfo->dqi_syncms = le32_to_cpu(gdinfo->dqi_syncms);
3965 		oinfo->dqi_gi.dqi_blocks = le32_to_cpu(gdinfo->dqi_blocks);
3966 		oinfo->dqi_gi.dqi_free_blk = le32_to_cpu(gdinfo->dqi_free_blk);
3967 		oinfo->dqi_gi.dqi_free_entry =
3968 					le32_to_cpu(gdinfo->dqi_free_entry);
3969 		brelse(bh);
3970 		ocfs2_track_lock_refresh(lockres);
3971 	}
3972 
3973 bail:
3974 	return status;
3975 }
3976 
3977 /* Lock quota info, this function expects at least shared lock on the quota file
3978  * so that we can safely refresh quota info from disk. */
3979 int ocfs2_qinfo_lock(struct ocfs2_mem_dqinfo *oinfo, int ex)
3980 {
3981 	struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
3982 	struct ocfs2_super *osb = OCFS2_SB(oinfo->dqi_gi.dqi_sb);
3983 	int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
3984 	int status = 0;
3985 
3986 	/* On RO devices, locking really isn't needed... */
3987 	if (ocfs2_is_hard_readonly(osb)) {
3988 		if (ex)
3989 			status = -EROFS;
3990 		goto bail;
3991 	}
3992 	if (ocfs2_mount_local(osb))
3993 		goto bail;
3994 
3995 	status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
3996 	if (status < 0) {
3997 		mlog_errno(status);
3998 		goto bail;
3999 	}
4000 	if (!ocfs2_should_refresh_lock_res(lockres))
4001 		goto bail;
4002 	/* OK, we have the lock but we need to refresh the quota info */
4003 	status = ocfs2_refresh_qinfo(oinfo);
4004 	if (status)
4005 		ocfs2_qinfo_unlock(oinfo, ex);
4006 	ocfs2_complete_lock_res_refresh(lockres, status);
4007 bail:
4008 	return status;
4009 }
4010 
4011 int ocfs2_refcount_lock(struct ocfs2_refcount_tree *ref_tree, int ex)
4012 {
4013 	int status;
4014 	int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4015 	struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
4016 	struct ocfs2_super *osb = lockres->l_priv;
4017 
4018 
4019 	if (ocfs2_is_hard_readonly(osb))
4020 		return -EROFS;
4021 
4022 	if (ocfs2_mount_local(osb))
4023 		return 0;
4024 
4025 	status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
4026 	if (status < 0)
4027 		mlog_errno(status);
4028 
4029 	return status;
4030 }
4031 
4032 void ocfs2_refcount_unlock(struct ocfs2_refcount_tree *ref_tree, int ex)
4033 {
4034 	int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4035 	struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
4036 	struct ocfs2_super *osb = lockres->l_priv;
4037 
4038 	if (!ocfs2_mount_local(osb))
4039 		ocfs2_cluster_unlock(osb, lockres, level);
4040 }
4041 
4042 static void ocfs2_process_blocked_lock(struct ocfs2_super *osb,
4043 				       struct ocfs2_lock_res *lockres)
4044 {
4045 	int status;
4046 	struct ocfs2_unblock_ctl ctl = {0, 0,};
4047 	unsigned long flags;
4048 
4049 	/* Our reference to the lockres in this function can be
4050 	 * considered valid until we remove the OCFS2_LOCK_QUEUED
4051 	 * flag. */
4052 
4053 	BUG_ON(!lockres);
4054 	BUG_ON(!lockres->l_ops);
4055 
4056 	mlog(ML_BASTS, "lockres %s blocked\n", lockres->l_name);
4057 
4058 	/* Detect whether a lock has been marked as going away while
4059 	 * the downconvert thread was processing other things. A lock can
4060 	 * still be marked with OCFS2_LOCK_FREEING after this check,
4061 	 * but short circuiting here will still save us some
4062 	 * performance. */
4063 	spin_lock_irqsave(&lockres->l_lock, flags);
4064 	if (lockres->l_flags & OCFS2_LOCK_FREEING)
4065 		goto unqueue;
4066 	spin_unlock_irqrestore(&lockres->l_lock, flags);
4067 
4068 	status = ocfs2_unblock_lock(osb, lockres, &ctl);
4069 	if (status < 0)
4070 		mlog_errno(status);
4071 
4072 	spin_lock_irqsave(&lockres->l_lock, flags);
4073 unqueue:
4074 	if (lockres->l_flags & OCFS2_LOCK_FREEING || !ctl.requeue) {
4075 		lockres_clear_flags(lockres, OCFS2_LOCK_QUEUED);
4076 	} else
4077 		ocfs2_schedule_blocked_lock(osb, lockres);
4078 
4079 	mlog(ML_BASTS, "lockres %s, requeue = %s.\n", lockres->l_name,
4080 	     ctl.requeue ? "yes" : "no");
4081 	spin_unlock_irqrestore(&lockres->l_lock, flags);
4082 
4083 	if (ctl.unblock_action != UNBLOCK_CONTINUE
4084 	    && lockres->l_ops->post_unlock)
4085 		lockres->l_ops->post_unlock(osb, lockres);
4086 }
4087 
4088 static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
4089 					struct ocfs2_lock_res *lockres)
4090 {
4091 	unsigned long flags;
4092 
4093 	assert_spin_locked(&lockres->l_lock);
4094 
4095 	if (lockres->l_flags & OCFS2_LOCK_FREEING) {
4096 		/* Do not schedule a lock for downconvert when it's on
4097 		 * the way to destruction - any nodes wanting access
4098 		 * to the resource will get it soon. */
4099 		mlog(ML_BASTS, "lockres %s won't be scheduled: flags 0x%lx\n",
4100 		     lockres->l_name, lockres->l_flags);
4101 		return;
4102 	}
4103 
4104 	lockres_or_flags(lockres, OCFS2_LOCK_QUEUED);
4105 
4106 	spin_lock_irqsave(&osb->dc_task_lock, flags);
4107 	if (list_empty(&lockres->l_blocked_list)) {
4108 		list_add_tail(&lockres->l_blocked_list,
4109 			      &osb->blocked_lock_list);
4110 		osb->blocked_lock_count++;
4111 	}
4112 	spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4113 }
4114 
4115 static void ocfs2_downconvert_thread_do_work(struct ocfs2_super *osb)
4116 {
4117 	unsigned long processed;
4118 	unsigned long flags;
4119 	struct ocfs2_lock_res *lockres;
4120 
4121 	spin_lock_irqsave(&osb->dc_task_lock, flags);
4122 	/* grab this early so we know to try again if a state change and
4123 	 * wake happens part-way through our work  */
4124 	osb->dc_work_sequence = osb->dc_wake_sequence;
4125 
4126 	processed = osb->blocked_lock_count;
4127 	/*
4128 	 * blocked lock processing in this loop might call iput which can
4129 	 * remove items off osb->blocked_lock_list. Downconvert up to
4130 	 * 'processed' number of locks, but stop short if we had some
4131 	 * removed in ocfs2_mark_lockres_freeing when downconverting.
4132 	 */
4133 	while (processed && !list_empty(&osb->blocked_lock_list)) {
4134 		lockres = list_entry(osb->blocked_lock_list.next,
4135 				     struct ocfs2_lock_res, l_blocked_list);
4136 		list_del_init(&lockres->l_blocked_list);
4137 		osb->blocked_lock_count--;
4138 		spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4139 
4140 		BUG_ON(!processed);
4141 		processed--;
4142 
4143 		ocfs2_process_blocked_lock(osb, lockres);
4144 
4145 		spin_lock_irqsave(&osb->dc_task_lock, flags);
4146 	}
4147 	spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4148 }
4149 
4150 static int ocfs2_downconvert_thread_lists_empty(struct ocfs2_super *osb)
4151 {
4152 	int empty = 0;
4153 	unsigned long flags;
4154 
4155 	spin_lock_irqsave(&osb->dc_task_lock, flags);
4156 	if (list_empty(&osb->blocked_lock_list))
4157 		empty = 1;
4158 
4159 	spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4160 	return empty;
4161 }
4162 
4163 static int ocfs2_downconvert_thread_should_wake(struct ocfs2_super *osb)
4164 {
4165 	int should_wake = 0;
4166 	unsigned long flags;
4167 
4168 	spin_lock_irqsave(&osb->dc_task_lock, flags);
4169 	if (osb->dc_work_sequence != osb->dc_wake_sequence)
4170 		should_wake = 1;
4171 	spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4172 
4173 	return should_wake;
4174 }
4175 
4176 static int ocfs2_downconvert_thread(void *arg)
4177 {
4178 	int status = 0;
4179 	struct ocfs2_super *osb = arg;
4180 
4181 	/* only quit once we've been asked to stop and there is no more
4182 	 * work available */
4183 	while (!(kthread_should_stop() &&
4184 		ocfs2_downconvert_thread_lists_empty(osb))) {
4185 
4186 		wait_event_interruptible(osb->dc_event,
4187 					 ocfs2_downconvert_thread_should_wake(osb) ||
4188 					 kthread_should_stop());
4189 
4190 		mlog(0, "downconvert_thread: awoken\n");
4191 
4192 		ocfs2_downconvert_thread_do_work(osb);
4193 	}
4194 
4195 	osb->dc_task = NULL;
4196 	return status;
4197 }
4198 
4199 void ocfs2_wake_downconvert_thread(struct ocfs2_super *osb)
4200 {
4201 	unsigned long flags;
4202 
4203 	spin_lock_irqsave(&osb->dc_task_lock, flags);
4204 	/* make sure the voting thread gets a swipe at whatever changes
4205 	 * the caller may have made to the voting state */
4206 	osb->dc_wake_sequence++;
4207 	spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4208 	wake_up(&osb->dc_event);
4209 }
4210