~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

TOMOYO Linux Cross Reference
Linux/fs/ocfs2/dlmglue.c

Version: ~ [ linux-5.2-rc1 ] ~ [ linux-5.1.2 ] ~ [ linux-5.0.16 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.43 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.119 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.176 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.179 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.139 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.67 ] ~ [ linux-3.15.10 ] ~ [ linux-3.14.79 ] ~ [ linux-3.13.11 ] ~ [ linux-3.12.74 ] ~ [ linux-3.11.10 ] ~ [ linux-3.10.108 ] ~ [ linux-3.9.11 ] ~ [ linux-3.8.13 ] ~ [ linux-3.7.10 ] ~ [ linux-3.6.11 ] ~ [ linux-3.5.7 ] ~ [ linux-3.4.113 ] ~ [ linux-3.3.8 ] ~ [ linux-3.2.102 ] ~ [ linux-3.1.10 ] ~ [ linux-3.0.101 ] ~ [ linux-2.6.39.4 ] ~ [ linux-2.6.38.8 ] ~ [ linux-2.6.37.6 ] ~ [ linux-2.6.36.4 ] ~ [ linux-2.6.35.14 ] ~ [ linux-2.6.34.15 ] ~ [ linux-2.6.33.20 ] ~ [ linux-2.6.32.71 ] ~ [ linux-2.6.0 ] ~ [ linux-2.4.37.11 ] ~ [ unix-v6-master ] ~ [ ccs-tools-1.8.5 ] ~ [ policy-sample ] ~
Architecture: ~ [ i386 ] ~ [ alpha ] ~ [ m68k ] ~ [ mips ] ~ [ ppc ] ~ [ sparc ] ~ [ sparc64 ] ~

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

~ [ source navigation ] ~ [ diff markup ] ~ [ identifier search ] ~

kernel.org | git.kernel.org | LWN.net | Project Home | Wiki (Japanese) | Wiki (English) | SVN repository | Mail admin

Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.

osdn.jp