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

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

Version: ~ [ linux-5.5-rc1 ] ~ [ linux-5.4.2 ] ~ [ linux-5.3.15 ] ~ [ linux-5.2.21 ] ~ [ linux-5.1.21 ] ~ [ linux-5.0.21 ] ~ [ linux-4.20.17 ] ~ [ linux-4.19.88 ] ~ [ linux-4.18.20 ] ~ [ linux-4.17.19 ] ~ [ linux-4.16.18 ] ~ [ linux-4.15.18 ] ~ [ linux-4.14.158 ] ~ [ linux-4.13.16 ] ~ [ linux-4.12.14 ] ~ [ linux-4.11.12 ] ~ [ linux-4.10.17 ] ~ [ linux-4.9.206 ] ~ [ linux-4.8.17 ] ~ [ linux-4.7.10 ] ~ [ linux-4.6.7 ] ~ [ linux-4.5.7 ] ~ [ linux-4.4.206 ] ~ [ linux-4.3.6 ] ~ [ linux-4.2.8 ] ~ [ linux-4.1.52 ] ~ [ linux-4.0.9 ] ~ [ linux-3.19.8 ] ~ [ linux-3.18.140 ] ~ [ linux-3.17.8 ] ~ [ linux-3.16.78 ] ~ [ 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.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(!inode);
1639         BUG_ON(!ocfs2_inode_is_new(inode));
1640 
1641         mlog(0, "Inode %llu\n", (unsigned long long)OCFS2_I(inode)->ip_blkno);
1642 
1643         /* NOTE: That we don't increment any of the holder counts, nor
1644          * do we add anything to a journal handle. Since this is
1645          * supposed to be a new inode which the cluster doesn't know
1646          * about yet, there is no need to.  As far as the LVB handling
1647          * is concerned, this is basically like acquiring an EX lock
1648          * on a resource which has an invalid one -- we'll set it
1649          * valid when we release the EX. */
1650 
1651         ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_rw_lockres, 1, 1);
1652         if (ret) {
1653                 mlog_errno(ret);
1654                 goto bail;
1655         }
1656 
1657         /*
1658          * We don't want to use DLM_LKF_LOCAL on a meta data lock as they
1659          * don't use a generation in their lock names.
1660          */
1661         ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_inode_lockres, 1, 0);
1662         if (ret) {
1663                 mlog_errno(ret);
1664                 goto bail;
1665         }
1666 
1667         ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_open_lockres, 0, 0);
1668         if (ret) {
1669                 mlog_errno(ret);
1670                 goto bail;
1671         }
1672 
1673 bail:
1674         return ret;
1675 }
1676 
1677 int ocfs2_rw_lock(struct inode *inode, int write)
1678 {
1679         int status, level;
1680         struct ocfs2_lock_res *lockres;
1681         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1682 
1683         BUG_ON(!inode);
1684 
1685         mlog(0, "inode %llu take %s RW lock\n",
1686              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1687              write ? "EXMODE" : "PRMODE");
1688 
1689         if (ocfs2_mount_local(osb))
1690                 return 0;
1691 
1692         lockres = &OCFS2_I(inode)->ip_rw_lockres;
1693 
1694         level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1695 
1696         status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres, level, 0,
1697                                     0);
1698         if (status < 0)
1699                 mlog_errno(status);
1700 
1701         return status;
1702 }
1703 
1704 void ocfs2_rw_unlock(struct inode *inode, int write)
1705 {
1706         int level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1707         struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_rw_lockres;
1708         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1709 
1710         mlog(0, "inode %llu drop %s RW lock\n",
1711              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1712              write ? "EXMODE" : "PRMODE");
1713 
1714         if (!ocfs2_mount_local(osb))
1715                 ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres, level);
1716 }
1717 
1718 /*
1719  * ocfs2_open_lock always get PR mode lock.
1720  */
1721 int ocfs2_open_lock(struct inode *inode)
1722 {
1723         int status = 0;
1724         struct ocfs2_lock_res *lockres;
1725         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1726 
1727         BUG_ON(!inode);
1728 
1729         mlog(0, "inode %llu take PRMODE open lock\n",
1730              (unsigned long long)OCFS2_I(inode)->ip_blkno);
1731 
1732         if (ocfs2_is_hard_readonly(osb) || ocfs2_mount_local(osb))
1733                 goto out;
1734 
1735         lockres = &OCFS2_I(inode)->ip_open_lockres;
1736 
1737         status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres,
1738                                     DLM_LOCK_PR, 0, 0);
1739         if (status < 0)
1740                 mlog_errno(status);
1741 
1742 out:
1743         return status;
1744 }
1745 
1746 int ocfs2_try_open_lock(struct inode *inode, int write)
1747 {
1748         int status = 0, level;
1749         struct ocfs2_lock_res *lockres;
1750         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1751 
1752         BUG_ON(!inode);
1753 
1754         mlog(0, "inode %llu try to take %s open lock\n",
1755              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1756              write ? "EXMODE" : "PRMODE");
1757 
1758         if (ocfs2_is_hard_readonly(osb)) {
1759                 if (write)
1760                         status = -EROFS;
1761                 goto out;
1762         }
1763 
1764         if (ocfs2_mount_local(osb))
1765                 goto out;
1766 
1767         lockres = &OCFS2_I(inode)->ip_open_lockres;
1768 
1769         level = write ? DLM_LOCK_EX : DLM_LOCK_PR;
1770 
1771         /*
1772          * The file system may already holding a PRMODE/EXMODE open lock.
1773          * Since we pass DLM_LKF_NOQUEUE, the request won't block waiting on
1774          * other nodes and the -EAGAIN will indicate to the caller that
1775          * this inode is still in use.
1776          */
1777         status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres,
1778                                     level, DLM_LKF_NOQUEUE, 0);
1779 
1780 out:
1781         return status;
1782 }
1783 
1784 /*
1785  * ocfs2_open_unlock unlock PR and EX mode open locks.
1786  */
1787 void ocfs2_open_unlock(struct inode *inode)
1788 {
1789         struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_open_lockres;
1790         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1791 
1792         mlog(0, "inode %llu drop open lock\n",
1793              (unsigned long long)OCFS2_I(inode)->ip_blkno);
1794 
1795         if (ocfs2_mount_local(osb))
1796                 goto out;
1797 
1798         if(lockres->l_ro_holders)
1799                 ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres,
1800                                      DLM_LOCK_PR);
1801         if(lockres->l_ex_holders)
1802                 ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres,
1803                                      DLM_LOCK_EX);
1804 
1805 out:
1806         return;
1807 }
1808 
1809 static int ocfs2_flock_handle_signal(struct ocfs2_lock_res *lockres,
1810                                      int level)
1811 {
1812         int ret;
1813         struct ocfs2_super *osb = ocfs2_get_lockres_osb(lockres);
1814         unsigned long flags;
1815         struct ocfs2_mask_waiter mw;
1816 
1817         ocfs2_init_mask_waiter(&mw);
1818 
1819 retry_cancel:
1820         spin_lock_irqsave(&lockres->l_lock, flags);
1821         if (lockres->l_flags & OCFS2_LOCK_BUSY) {
1822                 ret = ocfs2_prepare_cancel_convert(osb, lockres);
1823                 if (ret) {
1824                         spin_unlock_irqrestore(&lockres->l_lock, flags);
1825                         ret = ocfs2_cancel_convert(osb, lockres);
1826                         if (ret < 0) {
1827                                 mlog_errno(ret);
1828                                 goto out;
1829                         }
1830                         goto retry_cancel;
1831                 }
1832                 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1833                 spin_unlock_irqrestore(&lockres->l_lock, flags);
1834 
1835                 ocfs2_wait_for_mask(&mw);
1836                 goto retry_cancel;
1837         }
1838 
1839         ret = -ERESTARTSYS;
1840         /*
1841          * We may still have gotten the lock, in which case there's no
1842          * point to restarting the syscall.
1843          */
1844         if (lockres->l_level == level)
1845                 ret = 0;
1846 
1847         mlog(0, "Cancel returning %d. flags: 0x%lx, level: %d, act: %d\n", ret,
1848              lockres->l_flags, lockres->l_level, lockres->l_action);
1849 
1850         spin_unlock_irqrestore(&lockres->l_lock, flags);
1851 
1852 out:
1853         return ret;
1854 }
1855 
1856 /*
1857  * ocfs2_file_lock() and ocfs2_file_unlock() map to a single pair of
1858  * flock() calls. The locking approach this requires is sufficiently
1859  * different from all other cluster lock types that we implement a
1860  * separate path to the "low-level" dlm calls. In particular:
1861  *
1862  * - No optimization of lock levels is done - we take at exactly
1863  *   what's been requested.
1864  *
1865  * - No lock caching is employed. We immediately downconvert to
1866  *   no-lock at unlock time. This also means flock locks never go on
1867  *   the blocking list).
1868  *
1869  * - Since userspace can trivially deadlock itself with flock, we make
1870  *   sure to allow cancellation of a misbehaving applications flock()
1871  *   request.
1872  *
1873  * - Access to any flock lockres doesn't require concurrency, so we
1874  *   can simplify the code by requiring the caller to guarantee
1875  *   serialization of dlmglue flock calls.
1876  */
1877 int ocfs2_file_lock(struct file *file, int ex, int trylock)
1878 {
1879         int ret, level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
1880         unsigned int lkm_flags = trylock ? DLM_LKF_NOQUEUE : 0;
1881         unsigned long flags;
1882         struct ocfs2_file_private *fp = file->private_data;
1883         struct ocfs2_lock_res *lockres = &fp->fp_flock;
1884         struct ocfs2_super *osb = OCFS2_SB(file->f_mapping->host->i_sb);
1885         struct ocfs2_mask_waiter mw;
1886 
1887         ocfs2_init_mask_waiter(&mw);
1888 
1889         if ((lockres->l_flags & OCFS2_LOCK_BUSY) ||
1890             (lockres->l_level > DLM_LOCK_NL)) {
1891                 mlog(ML_ERROR,
1892                      "File lock \"%s\" has busy or locked state: flags: 0x%lx, "
1893                      "level: %u\n", lockres->l_name, lockres->l_flags,
1894                      lockres->l_level);
1895                 return -EINVAL;
1896         }
1897 
1898         spin_lock_irqsave(&lockres->l_lock, flags);
1899         if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
1900                 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1901                 spin_unlock_irqrestore(&lockres->l_lock, flags);
1902 
1903                 /*
1904                  * Get the lock at NLMODE to start - that way we
1905                  * can cancel the upconvert request if need be.
1906                  */
1907                 ret = ocfs2_lock_create(osb, lockres, DLM_LOCK_NL, 0);
1908                 if (ret < 0) {
1909                         mlog_errno(ret);
1910                         goto out;
1911                 }
1912 
1913                 ret = ocfs2_wait_for_mask(&mw);
1914                 if (ret) {
1915                         mlog_errno(ret);
1916                         goto out;
1917                 }
1918                 spin_lock_irqsave(&lockres->l_lock, flags);
1919         }
1920 
1921         lockres->l_action = OCFS2_AST_CONVERT;
1922         lkm_flags |= DLM_LKF_CONVERT;
1923         lockres->l_requested = level;
1924         lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
1925 
1926         lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
1927         spin_unlock_irqrestore(&lockres->l_lock, flags);
1928 
1929         ret = ocfs2_dlm_lock(osb->cconn, level, &lockres->l_lksb, lkm_flags,
1930                              lockres->l_name, OCFS2_LOCK_ID_MAX_LEN - 1);
1931         if (ret) {
1932                 if (!trylock || (ret != -EAGAIN)) {
1933                         ocfs2_log_dlm_error("ocfs2_dlm_lock", ret, lockres);
1934                         ret = -EINVAL;
1935                 }
1936 
1937                 ocfs2_recover_from_dlm_error(lockres, 1);
1938                 lockres_remove_mask_waiter(lockres, &mw);
1939                 goto out;
1940         }
1941 
1942         ret = ocfs2_wait_for_mask_interruptible(&mw, lockres);
1943         if (ret == -ERESTARTSYS) {
1944                 /*
1945                  * Userspace can cause deadlock itself with
1946                  * flock(). Current behavior locally is to allow the
1947                  * deadlock, but abort the system call if a signal is
1948                  * received. We follow this example, otherwise a
1949                  * poorly written program could sit in kernel until
1950                  * reboot.
1951                  *
1952                  * Handling this is a bit more complicated for Ocfs2
1953                  * though. We can't exit this function with an
1954                  * outstanding lock request, so a cancel convert is
1955                  * required. We intentionally overwrite 'ret' - if the
1956                  * cancel fails and the lock was granted, it's easier
1957                  * to just bubble success back up to the user.
1958                  */
1959                 ret = ocfs2_flock_handle_signal(lockres, level);
1960         } else if (!ret && (level > lockres->l_level)) {
1961                 /* Trylock failed asynchronously */
1962                 BUG_ON(!trylock);
1963                 ret = -EAGAIN;
1964         }
1965 
1966 out:
1967 
1968         mlog(0, "Lock: \"%s\" ex: %d, trylock: %d, returns: %d\n",
1969              lockres->l_name, ex, trylock, ret);
1970         return ret;
1971 }
1972 
1973 void ocfs2_file_unlock(struct file *file)
1974 {
1975         int ret;
1976         unsigned int gen;
1977         unsigned long flags;
1978         struct ocfs2_file_private *fp = file->private_data;
1979         struct ocfs2_lock_res *lockres = &fp->fp_flock;
1980         struct ocfs2_super *osb = OCFS2_SB(file->f_mapping->host->i_sb);
1981         struct ocfs2_mask_waiter mw;
1982 
1983         ocfs2_init_mask_waiter(&mw);
1984 
1985         if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED))
1986                 return;
1987 
1988         if (lockres->l_level == DLM_LOCK_NL)
1989                 return;
1990 
1991         mlog(0, "Unlock: \"%s\" flags: 0x%lx, level: %d, act: %d\n",
1992              lockres->l_name, lockres->l_flags, lockres->l_level,
1993              lockres->l_action);
1994 
1995         spin_lock_irqsave(&lockres->l_lock, flags);
1996         /*
1997          * Fake a blocking ast for the downconvert code.
1998          */
1999         lockres_or_flags(lockres, OCFS2_LOCK_BLOCKED);
2000         lockres->l_blocking = DLM_LOCK_EX;
2001 
2002         gen = ocfs2_prepare_downconvert(lockres, DLM_LOCK_NL);
2003         lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
2004         spin_unlock_irqrestore(&lockres->l_lock, flags);
2005 
2006         ret = ocfs2_downconvert_lock(osb, lockres, DLM_LOCK_NL, 0, gen);
2007         if (ret) {
2008                 mlog_errno(ret);
2009                 return;
2010         }
2011 
2012         ret = ocfs2_wait_for_mask(&mw);
2013         if (ret)
2014                 mlog_errno(ret);
2015 }
2016 
2017 static void ocfs2_downconvert_on_unlock(struct ocfs2_super *osb,
2018                                         struct ocfs2_lock_res *lockres)
2019 {
2020         int kick = 0;
2021 
2022         /* If we know that another node is waiting on our lock, kick
2023          * the downconvert thread * pre-emptively when we reach a release
2024          * condition. */
2025         if (lockres->l_flags & OCFS2_LOCK_BLOCKED) {
2026                 switch(lockres->l_blocking) {
2027                 case DLM_LOCK_EX:
2028                         if (!lockres->l_ex_holders && !lockres->l_ro_holders)
2029                                 kick = 1;
2030                         break;
2031                 case DLM_LOCK_PR:
2032                         if (!lockres->l_ex_holders)
2033                                 kick = 1;
2034                         break;
2035                 default:
2036                         BUG();
2037                 }
2038         }
2039 
2040         if (kick)
2041                 ocfs2_wake_downconvert_thread(osb);
2042 }
2043 
2044 #define OCFS2_SEC_BITS   34
2045 #define OCFS2_SEC_SHIFT  (64 - 34)
2046 #define OCFS2_NSEC_MASK  ((1ULL << OCFS2_SEC_SHIFT) - 1)
2047 
2048 /* LVB only has room for 64 bits of time here so we pack it for
2049  * now. */
2050 static u64 ocfs2_pack_timespec(struct timespec *spec)
2051 {
2052         u64 res;
2053         u64 sec = spec->tv_sec;
2054         u32 nsec = spec->tv_nsec;
2055 
2056         res = (sec << OCFS2_SEC_SHIFT) | (nsec & OCFS2_NSEC_MASK);
2057 
2058         return res;
2059 }
2060 
2061 /* Call this with the lockres locked. I am reasonably sure we don't
2062  * need ip_lock in this function as anyone who would be changing those
2063  * values is supposed to be blocked in ocfs2_inode_lock right now. */
2064 static void __ocfs2_stuff_meta_lvb(struct inode *inode)
2065 {
2066         struct ocfs2_inode_info *oi = OCFS2_I(inode);
2067         struct ocfs2_lock_res *lockres = &oi->ip_inode_lockres;
2068         struct ocfs2_meta_lvb *lvb;
2069 
2070         lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2071 
2072         /*
2073          * Invalidate the LVB of a deleted inode - this way other
2074          * nodes are forced to go to disk and discover the new inode
2075          * status.
2076          */
2077         if (oi->ip_flags & OCFS2_INODE_DELETED) {
2078                 lvb->lvb_version = 0;
2079                 goto out;
2080         }
2081 
2082         lvb->lvb_version   = OCFS2_LVB_VERSION;
2083         lvb->lvb_isize     = cpu_to_be64(i_size_read(inode));
2084         lvb->lvb_iclusters = cpu_to_be32(oi->ip_clusters);
2085         lvb->lvb_iuid      = cpu_to_be32(i_uid_read(inode));
2086         lvb->lvb_igid      = cpu_to_be32(i_gid_read(inode));
2087         lvb->lvb_imode     = cpu_to_be16(inode->i_mode);
2088         lvb->lvb_inlink    = cpu_to_be16(inode->i_nlink);
2089         lvb->lvb_iatime_packed  =
2090                 cpu_to_be64(ocfs2_pack_timespec(&inode->i_atime));
2091         lvb->lvb_ictime_packed =
2092                 cpu_to_be64(ocfs2_pack_timespec(&inode->i_ctime));
2093         lvb->lvb_imtime_packed =
2094                 cpu_to_be64(ocfs2_pack_timespec(&inode->i_mtime));
2095         lvb->lvb_iattr    = cpu_to_be32(oi->ip_attr);
2096         lvb->lvb_idynfeatures = cpu_to_be16(oi->ip_dyn_features);
2097         lvb->lvb_igeneration = cpu_to_be32(inode->i_generation);
2098 
2099 out:
2100         mlog_meta_lvb(0, lockres);
2101 }
2102 
2103 static void ocfs2_unpack_timespec(struct timespec *spec,
2104                                   u64 packed_time)
2105 {
2106         spec->tv_sec = packed_time >> OCFS2_SEC_SHIFT;
2107         spec->tv_nsec = packed_time & OCFS2_NSEC_MASK;
2108 }
2109 
2110 static void ocfs2_refresh_inode_from_lvb(struct inode *inode)
2111 {
2112         struct ocfs2_inode_info *oi = OCFS2_I(inode);
2113         struct ocfs2_lock_res *lockres = &oi->ip_inode_lockres;
2114         struct ocfs2_meta_lvb *lvb;
2115 
2116         mlog_meta_lvb(0, lockres);
2117 
2118         lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2119 
2120         /* We're safe here without the lockres lock... */
2121         spin_lock(&oi->ip_lock);
2122         oi->ip_clusters = be32_to_cpu(lvb->lvb_iclusters);
2123         i_size_write(inode, be64_to_cpu(lvb->lvb_isize));
2124 
2125         oi->ip_attr = be32_to_cpu(lvb->lvb_iattr);
2126         oi->ip_dyn_features = be16_to_cpu(lvb->lvb_idynfeatures);
2127         ocfs2_set_inode_flags(inode);
2128 
2129         /* fast-symlinks are a special case */
2130         if (S_ISLNK(inode->i_mode) && !oi->ip_clusters)
2131                 inode->i_blocks = 0;
2132         else
2133                 inode->i_blocks = ocfs2_inode_sector_count(inode);
2134 
2135         i_uid_write(inode, be32_to_cpu(lvb->lvb_iuid));
2136         i_gid_write(inode, be32_to_cpu(lvb->lvb_igid));
2137         inode->i_mode    = be16_to_cpu(lvb->lvb_imode);
2138         set_nlink(inode, be16_to_cpu(lvb->lvb_inlink));
2139         ocfs2_unpack_timespec(&inode->i_atime,
2140                               be64_to_cpu(lvb->lvb_iatime_packed));
2141         ocfs2_unpack_timespec(&inode->i_mtime,
2142                               be64_to_cpu(lvb->lvb_imtime_packed));
2143         ocfs2_unpack_timespec(&inode->i_ctime,
2144                               be64_to_cpu(lvb->lvb_ictime_packed));
2145         spin_unlock(&oi->ip_lock);
2146 }
2147 
2148 static inline int ocfs2_meta_lvb_is_trustable(struct inode *inode,
2149                                               struct ocfs2_lock_res *lockres)
2150 {
2151         struct ocfs2_meta_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2152 
2153         if (ocfs2_dlm_lvb_valid(&lockres->l_lksb)
2154             && lvb->lvb_version == OCFS2_LVB_VERSION
2155             && be32_to_cpu(lvb->lvb_igeneration) == inode->i_generation)
2156                 return 1;
2157         return 0;
2158 }
2159 
2160 /* Determine whether a lock resource needs to be refreshed, and
2161  * arbitrate who gets to refresh it.
2162  *
2163  *   0 means no refresh needed.
2164  *
2165  *   > 0 means you need to refresh this and you MUST call
2166  *   ocfs2_complete_lock_res_refresh afterwards. */
2167 static int ocfs2_should_refresh_lock_res(struct ocfs2_lock_res *lockres)
2168 {
2169         unsigned long flags;
2170         int status = 0;
2171 
2172 refresh_check:
2173         spin_lock_irqsave(&lockres->l_lock, flags);
2174         if (!(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH)) {
2175                 spin_unlock_irqrestore(&lockres->l_lock, flags);
2176                 goto bail;
2177         }
2178 
2179         if (lockres->l_flags & OCFS2_LOCK_REFRESHING) {
2180                 spin_unlock_irqrestore(&lockres->l_lock, flags);
2181 
2182                 ocfs2_wait_on_refreshing_lock(lockres);
2183                 goto refresh_check;
2184         }
2185 
2186         /* Ok, I'll be the one to refresh this lock. */
2187         lockres_or_flags(lockres, OCFS2_LOCK_REFRESHING);
2188         spin_unlock_irqrestore(&lockres->l_lock, flags);
2189 
2190         status = 1;
2191 bail:
2192         mlog(0, "status %d\n", status);
2193         return status;
2194 }
2195 
2196 /* If status is non zero, I'll mark it as not being in refresh
2197  * anymroe, but i won't clear the needs refresh flag. */
2198 static inline void ocfs2_complete_lock_res_refresh(struct ocfs2_lock_res *lockres,
2199                                                    int status)
2200 {
2201         unsigned long flags;
2202 
2203         spin_lock_irqsave(&lockres->l_lock, flags);
2204         lockres_clear_flags(lockres, OCFS2_LOCK_REFRESHING);
2205         if (!status)
2206                 lockres_clear_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
2207         spin_unlock_irqrestore(&lockres->l_lock, flags);
2208 
2209         wake_up(&lockres->l_event);
2210 }
2211 
2212 /* may or may not return a bh if it went to disk. */
2213 static int ocfs2_inode_lock_update(struct inode *inode,
2214                                   struct buffer_head **bh)
2215 {
2216         int status = 0;
2217         struct ocfs2_inode_info *oi = OCFS2_I(inode);
2218         struct ocfs2_lock_res *lockres = &oi->ip_inode_lockres;
2219         struct ocfs2_dinode *fe;
2220         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
2221 
2222         if (ocfs2_mount_local(osb))
2223                 goto bail;
2224 
2225         spin_lock(&oi->ip_lock);
2226         if (oi->ip_flags & OCFS2_INODE_DELETED) {
2227                 mlog(0, "Orphaned inode %llu was deleted while we "
2228                      "were waiting on a lock. ip_flags = 0x%x\n",
2229                      (unsigned long long)oi->ip_blkno, oi->ip_flags);
2230                 spin_unlock(&oi->ip_lock);
2231                 status = -ENOENT;
2232                 goto bail;
2233         }
2234         spin_unlock(&oi->ip_lock);
2235 
2236         if (!ocfs2_should_refresh_lock_res(lockres))
2237                 goto bail;
2238 
2239         /* This will discard any caching information we might have had
2240          * for the inode metadata. */
2241         ocfs2_metadata_cache_purge(INODE_CACHE(inode));
2242 
2243         ocfs2_extent_map_trunc(inode, 0);
2244 
2245         if (ocfs2_meta_lvb_is_trustable(inode, lockres)) {
2246                 mlog(0, "Trusting LVB on inode %llu\n",
2247                      (unsigned long long)oi->ip_blkno);
2248                 ocfs2_refresh_inode_from_lvb(inode);
2249         } else {
2250                 /* Boo, we have to go to disk. */
2251                 /* read bh, cast, ocfs2_refresh_inode */
2252                 status = ocfs2_read_inode_block(inode, bh);
2253                 if (status < 0) {
2254                         mlog_errno(status);
2255                         goto bail_refresh;
2256                 }
2257                 fe = (struct ocfs2_dinode *) (*bh)->b_data;
2258 
2259                 /* This is a good chance to make sure we're not
2260                  * locking an invalid object.  ocfs2_read_inode_block()
2261                  * already checked that the inode block is sane.
2262                  *
2263                  * We bug on a stale inode here because we checked
2264                  * above whether it was wiped from disk. The wiping
2265                  * node provides a guarantee that we receive that
2266                  * message and can mark the inode before dropping any
2267                  * locks associated with it. */
2268                 mlog_bug_on_msg(inode->i_generation !=
2269                                 le32_to_cpu(fe->i_generation),
2270                                 "Invalid dinode %llu disk generation: %u "
2271                                 "inode->i_generation: %u\n",
2272                                 (unsigned long long)oi->ip_blkno,
2273                                 le32_to_cpu(fe->i_generation),
2274                                 inode->i_generation);
2275                 mlog_bug_on_msg(le64_to_cpu(fe->i_dtime) ||
2276                                 !(fe->i_flags & cpu_to_le32(OCFS2_VALID_FL)),
2277                                 "Stale dinode %llu dtime: %llu flags: 0x%x\n",
2278                                 (unsigned long long)oi->ip_blkno,
2279                                 (unsigned long long)le64_to_cpu(fe->i_dtime),
2280                                 le32_to_cpu(fe->i_flags));
2281 
2282                 ocfs2_refresh_inode(inode, fe);
2283                 ocfs2_track_lock_refresh(lockres);
2284         }
2285 
2286         status = 0;
2287 bail_refresh:
2288         ocfs2_complete_lock_res_refresh(lockres, status);
2289 bail:
2290         return status;
2291 }
2292 
2293 static int ocfs2_assign_bh(struct inode *inode,
2294                            struct buffer_head **ret_bh,
2295                            struct buffer_head *passed_bh)
2296 {
2297         int status;
2298 
2299         if (passed_bh) {
2300                 /* Ok, the update went to disk for us, use the
2301                  * returned bh. */
2302                 *ret_bh = passed_bh;
2303                 get_bh(*ret_bh);
2304 
2305                 return 0;
2306         }
2307 
2308         status = ocfs2_read_inode_block(inode, ret_bh);
2309         if (status < 0)
2310                 mlog_errno(status);
2311 
2312         return status;
2313 }
2314 
2315 /*
2316  * returns < 0 error if the callback will never be called, otherwise
2317  * the result of the lock will be communicated via the callback.
2318  */
2319 int ocfs2_inode_lock_full_nested(struct inode *inode,
2320                                  struct buffer_head **ret_bh,
2321                                  int ex,
2322                                  int arg_flags,
2323                                  int subclass)
2324 {
2325         int status, level, acquired;
2326         u32 dlm_flags;
2327         struct ocfs2_lock_res *lockres = NULL;
2328         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
2329         struct buffer_head *local_bh = NULL;
2330 
2331         BUG_ON(!inode);
2332 
2333         mlog(0, "inode %llu, take %s META lock\n",
2334              (unsigned long long)OCFS2_I(inode)->ip_blkno,
2335              ex ? "EXMODE" : "PRMODE");
2336 
2337         status = 0;
2338         acquired = 0;
2339         /* We'll allow faking a readonly metadata lock for
2340          * rodevices. */
2341         if (ocfs2_is_hard_readonly(osb)) {
2342                 if (ex)
2343                         status = -EROFS;
2344                 goto getbh;
2345         }
2346 
2347         if (ocfs2_mount_local(osb))
2348                 goto local;
2349 
2350         if (!(arg_flags & OCFS2_META_LOCK_RECOVERY))
2351                 ocfs2_wait_for_recovery(osb);
2352 
2353         lockres = &OCFS2_I(inode)->ip_inode_lockres;
2354         level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2355         dlm_flags = 0;
2356         if (arg_flags & OCFS2_META_LOCK_NOQUEUE)
2357                 dlm_flags |= DLM_LKF_NOQUEUE;
2358 
2359         status = __ocfs2_cluster_lock(osb, lockres, level, dlm_flags,
2360                                       arg_flags, subclass, _RET_IP_);
2361         if (status < 0) {
2362                 if (status != -EAGAIN)
2363                         mlog_errno(status);
2364                 goto bail;
2365         }
2366 
2367         /* Notify the error cleanup path to drop the cluster lock. */
2368         acquired = 1;
2369 
2370         /* We wait twice because a node may have died while we were in
2371          * the lower dlm layers. The second time though, we've
2372          * committed to owning this lock so we don't allow signals to
2373          * abort the operation. */
2374         if (!(arg_flags & OCFS2_META_LOCK_RECOVERY))
2375                 ocfs2_wait_for_recovery(osb);
2376 
2377 local:
2378         /*
2379          * We only see this flag if we're being called from
2380          * ocfs2_read_locked_inode(). It means we're locking an inode
2381          * which hasn't been populated yet, so clear the refresh flag
2382          * and let the caller handle it.
2383          */
2384         if (inode->i_state & I_NEW) {
2385                 status = 0;
2386                 if (lockres)
2387                         ocfs2_complete_lock_res_refresh(lockres, 0);
2388                 goto bail;
2389         }
2390 
2391         /* This is fun. The caller may want a bh back, or it may
2392          * not. ocfs2_inode_lock_update definitely wants one in, but
2393          * may or may not read one, depending on what's in the
2394          * LVB. The result of all of this is that we've *only* gone to
2395          * disk if we have to, so the complexity is worthwhile. */
2396         status = ocfs2_inode_lock_update(inode, &local_bh);
2397         if (status < 0) {
2398                 if (status != -ENOENT)
2399                         mlog_errno(status);
2400                 goto bail;
2401         }
2402 getbh:
2403         if (ret_bh) {
2404                 status = ocfs2_assign_bh(inode, ret_bh, local_bh);
2405                 if (status < 0) {
2406                         mlog_errno(status);
2407                         goto bail;
2408                 }
2409         }
2410 
2411 bail:
2412         if (status < 0) {
2413                 if (ret_bh && (*ret_bh)) {
2414                         brelse(*ret_bh);
2415                         *ret_bh = NULL;
2416                 }
2417                 if (acquired)
2418                         ocfs2_inode_unlock(inode, ex);
2419         }
2420 
2421         if (local_bh)
2422                 brelse(local_bh);
2423 
2424         return status;
2425 }
2426 
2427 /*
2428  * This is working around a lock inversion between tasks acquiring DLM
2429  * locks while holding a page lock and the downconvert thread which
2430  * blocks dlm lock acquiry while acquiring page locks.
2431  *
2432  * ** These _with_page variantes are only intended to be called from aop
2433  * methods that hold page locks and return a very specific *positive* error
2434  * code that aop methods pass up to the VFS -- test for errors with != 0. **
2435  *
2436  * The DLM is called such that it returns -EAGAIN if it would have
2437  * blocked waiting for the downconvert thread.  In that case we unlock
2438  * our page so the downconvert thread can make progress.  Once we've
2439  * done this we have to return AOP_TRUNCATED_PAGE so the aop method
2440  * that called us can bubble that back up into the VFS who will then
2441  * immediately retry the aop call.
2442  */
2443 int ocfs2_inode_lock_with_page(struct inode *inode,
2444                               struct buffer_head **ret_bh,
2445                               int ex,
2446                               struct page *page)
2447 {
2448         int ret;
2449 
2450         ret = ocfs2_inode_lock_full(inode, ret_bh, ex, OCFS2_LOCK_NONBLOCK);
2451         if (ret == -EAGAIN) {
2452                 unlock_page(page);
2453                 ret = AOP_TRUNCATED_PAGE;
2454         }
2455 
2456         return ret;
2457 }
2458 
2459 int ocfs2_inode_lock_atime(struct inode *inode,
2460                           struct vfsmount *vfsmnt,
2461                           int *level)
2462 {
2463         int ret;
2464 
2465         ret = ocfs2_inode_lock(inode, NULL, 0);
2466         if (ret < 0) {
2467                 mlog_errno(ret);
2468                 return ret;
2469         }
2470 
2471         /*
2472          * If we should update atime, we will get EX lock,
2473          * otherwise we just get PR lock.
2474          */
2475         if (ocfs2_should_update_atime(inode, vfsmnt)) {
2476                 struct buffer_head *bh = NULL;
2477 
2478                 ocfs2_inode_unlock(inode, 0);
2479                 ret = ocfs2_inode_lock(inode, &bh, 1);
2480                 if (ret < 0) {
2481                         mlog_errno(ret);
2482                         return ret;
2483                 }
2484                 *level = 1;
2485                 if (ocfs2_should_update_atime(inode, vfsmnt))
2486                         ocfs2_update_inode_atime(inode, bh);
2487                 if (bh)
2488                         brelse(bh);
2489         } else
2490                 *level = 0;
2491 
2492         return ret;
2493 }
2494 
2495 void ocfs2_inode_unlock(struct inode *inode,
2496                        int ex)
2497 {
2498         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2499         struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_inode_lockres;
2500         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
2501 
2502         mlog(0, "inode %llu drop %s META lock\n",
2503              (unsigned long long)OCFS2_I(inode)->ip_blkno,
2504              ex ? "EXMODE" : "PRMODE");
2505 
2506         if (!ocfs2_is_hard_readonly(OCFS2_SB(inode->i_sb)) &&
2507             !ocfs2_mount_local(osb))
2508                 ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres, level);
2509 }
2510 
2511 int ocfs2_orphan_scan_lock(struct ocfs2_super *osb, u32 *seqno)
2512 {
2513         struct ocfs2_lock_res *lockres;
2514         struct ocfs2_orphan_scan_lvb *lvb;
2515         int status = 0;
2516 
2517         if (ocfs2_is_hard_readonly(osb))
2518                 return -EROFS;
2519 
2520         if (ocfs2_mount_local(osb))
2521                 return 0;
2522 
2523         lockres = &osb->osb_orphan_scan.os_lockres;
2524         status = ocfs2_cluster_lock(osb, lockres, DLM_LOCK_EX, 0, 0);
2525         if (status < 0)
2526                 return status;
2527 
2528         lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2529         if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
2530             lvb->lvb_version == OCFS2_ORPHAN_LVB_VERSION)
2531                 *seqno = be32_to_cpu(lvb->lvb_os_seqno);
2532         else
2533                 *seqno = osb->osb_orphan_scan.os_seqno + 1;
2534 
2535         return status;
2536 }
2537 
2538 void ocfs2_orphan_scan_unlock(struct ocfs2_super *osb, u32 seqno)
2539 {
2540         struct ocfs2_lock_res *lockres;
2541         struct ocfs2_orphan_scan_lvb *lvb;
2542 
2543         if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb)) {
2544                 lockres = &osb->osb_orphan_scan.os_lockres;
2545                 lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2546                 lvb->lvb_version = OCFS2_ORPHAN_LVB_VERSION;
2547                 lvb->lvb_os_seqno = cpu_to_be32(seqno);
2548                 ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
2549         }
2550 }
2551 
2552 int ocfs2_super_lock(struct ocfs2_super *osb,
2553                      int ex)
2554 {
2555         int status = 0;
2556         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2557         struct ocfs2_lock_res *lockres = &osb->osb_super_lockres;
2558 
2559         if (ocfs2_is_hard_readonly(osb))
2560                 return -EROFS;
2561 
2562         if (ocfs2_mount_local(osb))
2563                 goto bail;
2564 
2565         status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
2566         if (status < 0) {
2567                 mlog_errno(status);
2568                 goto bail;
2569         }
2570 
2571         /* The super block lock path is really in the best position to
2572          * know when resources covered by the lock need to be
2573          * refreshed, so we do it here. Of course, making sense of
2574          * everything is up to the caller :) */
2575         status = ocfs2_should_refresh_lock_res(lockres);
2576         if (status) {
2577                 status = ocfs2_refresh_slot_info(osb);
2578 
2579                 ocfs2_complete_lock_res_refresh(lockres, status);
2580 
2581                 if (status < 0) {
2582                         ocfs2_cluster_unlock(osb, lockres, level);
2583                         mlog_errno(status);
2584                 }
2585                 ocfs2_track_lock_refresh(lockres);
2586         }
2587 bail:
2588         return status;
2589 }
2590 
2591 void ocfs2_super_unlock(struct ocfs2_super *osb,
2592                         int ex)
2593 {
2594         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2595         struct ocfs2_lock_res *lockres = &osb->osb_super_lockres;
2596 
2597         if (!ocfs2_mount_local(osb))
2598                 ocfs2_cluster_unlock(osb, lockres, level);
2599 }
2600 
2601 int ocfs2_rename_lock(struct ocfs2_super *osb)
2602 {
2603         int status;
2604         struct ocfs2_lock_res *lockres = &osb->osb_rename_lockres;
2605 
2606         if (ocfs2_is_hard_readonly(osb))
2607                 return -EROFS;
2608 
2609         if (ocfs2_mount_local(osb))
2610                 return 0;
2611 
2612         status = ocfs2_cluster_lock(osb, lockres, DLM_LOCK_EX, 0, 0);
2613         if (status < 0)
2614                 mlog_errno(status);
2615 
2616         return status;
2617 }
2618 
2619 void ocfs2_rename_unlock(struct ocfs2_super *osb)
2620 {
2621         struct ocfs2_lock_res *lockres = &osb->osb_rename_lockres;
2622 
2623         if (!ocfs2_mount_local(osb))
2624                 ocfs2_cluster_unlock(osb, lockres, DLM_LOCK_EX);
2625 }
2626 
2627 int ocfs2_nfs_sync_lock(struct ocfs2_super *osb, int ex)
2628 {
2629         int status;
2630         struct ocfs2_lock_res *lockres = &osb->osb_nfs_sync_lockres;
2631 
2632         if (ocfs2_is_hard_readonly(osb))
2633                 return -EROFS;
2634 
2635         if (ocfs2_mount_local(osb))
2636                 return 0;
2637 
2638         status = ocfs2_cluster_lock(osb, lockres, ex ? LKM_EXMODE : LKM_PRMODE,
2639                                     0, 0);
2640         if (status < 0)
2641                 mlog(ML_ERROR, "lock on nfs sync lock failed %d\n", status);
2642 
2643         return status;
2644 }
2645 
2646 void ocfs2_nfs_sync_unlock(struct ocfs2_super *osb, int ex)
2647 {
2648         struct ocfs2_lock_res *lockres = &osb->osb_nfs_sync_lockres;
2649 
2650         if (!ocfs2_mount_local(osb))
2651                 ocfs2_cluster_unlock(osb, lockres,
2652                                      ex ? LKM_EXMODE : LKM_PRMODE);
2653 }
2654 
2655 int ocfs2_dentry_lock(struct dentry *dentry, int ex)
2656 {
2657         int ret;
2658         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2659         struct ocfs2_dentry_lock *dl = dentry->d_fsdata;
2660         struct ocfs2_super *osb = OCFS2_SB(dentry->d_sb);
2661 
2662         BUG_ON(!dl);
2663 
2664         if (ocfs2_is_hard_readonly(osb)) {
2665                 if (ex)
2666                         return -EROFS;
2667                 return 0;
2668         }
2669 
2670         if (ocfs2_mount_local(osb))
2671                 return 0;
2672 
2673         ret = ocfs2_cluster_lock(osb, &dl->dl_lockres, level, 0, 0);
2674         if (ret < 0)
2675                 mlog_errno(ret);
2676 
2677         return ret;
2678 }
2679 
2680 void ocfs2_dentry_unlock(struct dentry *dentry, int ex)
2681 {
2682         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
2683         struct ocfs2_dentry_lock *dl = dentry->d_fsdata;
2684         struct ocfs2_super *osb = OCFS2_SB(dentry->d_sb);
2685 
2686         if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb))
2687                 ocfs2_cluster_unlock(osb, &dl->dl_lockres, level);
2688 }
2689 
2690 /* Reference counting of the dlm debug structure. We want this because
2691  * open references on the debug inodes can live on after a mount, so
2692  * we can't rely on the ocfs2_super to always exist. */
2693 static void ocfs2_dlm_debug_free(struct kref *kref)
2694 {
2695         struct ocfs2_dlm_debug *dlm_debug;
2696 
2697         dlm_debug = container_of(kref, struct ocfs2_dlm_debug, d_refcnt);
2698 
2699         kfree(dlm_debug);
2700 }
2701 
2702 void ocfs2_put_dlm_debug(struct ocfs2_dlm_debug *dlm_debug)
2703 {
2704         if (dlm_debug)
2705                 kref_put(&dlm_debug->d_refcnt, ocfs2_dlm_debug_free);
2706 }
2707 
2708 static void ocfs2_get_dlm_debug(struct ocfs2_dlm_debug *debug)
2709 {
2710         kref_get(&debug->d_refcnt);
2711 }
2712 
2713 struct ocfs2_dlm_debug *ocfs2_new_dlm_debug(void)
2714 {
2715         struct ocfs2_dlm_debug *dlm_debug;
2716 
2717         dlm_debug = kmalloc(sizeof(struct ocfs2_dlm_debug), GFP_KERNEL);
2718         if (!dlm_debug) {
2719                 mlog_errno(-ENOMEM);
2720                 goto out;
2721         }
2722 
2723         kref_init(&dlm_debug->d_refcnt);
2724         INIT_LIST_HEAD(&dlm_debug->d_lockres_tracking);
2725         dlm_debug->d_locking_state = NULL;
2726 out:
2727         return dlm_debug;
2728 }
2729 
2730 /* Access to this is arbitrated for us via seq_file->sem. */
2731 struct ocfs2_dlm_seq_priv {
2732         struct ocfs2_dlm_debug *p_dlm_debug;
2733         struct ocfs2_lock_res p_iter_res;
2734         struct ocfs2_lock_res p_tmp_res;
2735 };
2736 
2737 static struct ocfs2_lock_res *ocfs2_dlm_next_res(struct ocfs2_lock_res *start,
2738                                                  struct ocfs2_dlm_seq_priv *priv)
2739 {
2740         struct ocfs2_lock_res *iter, *ret = NULL;
2741         struct ocfs2_dlm_debug *dlm_debug = priv->p_dlm_debug;
2742 
2743         assert_spin_locked(&ocfs2_dlm_tracking_lock);
2744 
2745         list_for_each_entry(iter, &start->l_debug_list, l_debug_list) {
2746                 /* discover the head of the list */
2747                 if (&iter->l_debug_list == &dlm_debug->d_lockres_tracking) {
2748                         mlog(0, "End of list found, %p\n", ret);
2749                         break;
2750                 }
2751 
2752                 /* We track our "dummy" iteration lockres' by a NULL
2753                  * l_ops field. */
2754                 if (iter->l_ops != NULL) {
2755                         ret = iter;
2756                         break;
2757                 }
2758         }
2759 
2760         return ret;
2761 }
2762 
2763 static void *ocfs2_dlm_seq_start(struct seq_file *m, loff_t *pos)
2764 {
2765         struct ocfs2_dlm_seq_priv *priv = m->private;
2766         struct ocfs2_lock_res *iter;
2767 
2768         spin_lock(&ocfs2_dlm_tracking_lock);
2769         iter = ocfs2_dlm_next_res(&priv->p_iter_res, priv);
2770         if (iter) {
2771                 /* Since lockres' have the lifetime of their container
2772                  * (which can be inodes, ocfs2_supers, etc) we want to
2773                  * copy this out to a temporary lockres while still
2774                  * under the spinlock. Obviously after this we can't
2775                  * trust any pointers on the copy returned, but that's
2776                  * ok as the information we want isn't typically held
2777                  * in them. */
2778                 priv->p_tmp_res = *iter;
2779                 iter = &priv->p_tmp_res;
2780         }
2781         spin_unlock(&ocfs2_dlm_tracking_lock);
2782 
2783         return iter;
2784 }
2785 
2786 static void ocfs2_dlm_seq_stop(struct seq_file *m, void *v)
2787 {
2788 }
2789 
2790 static void *ocfs2_dlm_seq_next(struct seq_file *m, void *v, loff_t *pos)
2791 {
2792         struct ocfs2_dlm_seq_priv *priv = m->private;
2793         struct ocfs2_lock_res *iter = v;
2794         struct ocfs2_lock_res *dummy = &priv->p_iter_res;
2795 
2796         spin_lock(&ocfs2_dlm_tracking_lock);
2797         iter = ocfs2_dlm_next_res(iter, priv);
2798         list_del_init(&dummy->l_debug_list);
2799         if (iter) {
2800                 list_add(&dummy->l_debug_list, &iter->l_debug_list);
2801                 priv->p_tmp_res = *iter;
2802                 iter = &priv->p_tmp_res;
2803         }
2804         spin_unlock(&ocfs2_dlm_tracking_lock);
2805 
2806         return iter;
2807 }
2808 
2809 /*
2810  * Version is used by debugfs.ocfs2 to determine the format being used
2811  *
2812  * New in version 2
2813  *      - Lock stats printed
2814  * New in version 3
2815  *      - Max time in lock stats is in usecs (instead of nsecs)
2816  */
2817 #define OCFS2_DLM_DEBUG_STR_VERSION 3
2818 static int ocfs2_dlm_seq_show(struct seq_file *m, void *v)
2819 {
2820         int i;
2821         char *lvb;
2822         struct ocfs2_lock_res *lockres = v;
2823 
2824         if (!lockres)
2825                 return -EINVAL;
2826 
2827         seq_printf(m, "0x%x\t", OCFS2_DLM_DEBUG_STR_VERSION);
2828 
2829         if (lockres->l_type == OCFS2_LOCK_TYPE_DENTRY)
2830                 seq_printf(m, "%.*s%08x\t", OCFS2_DENTRY_LOCK_INO_START - 1,
2831                            lockres->l_name,
2832                            (unsigned int)ocfs2_get_dentry_lock_ino(lockres));
2833         else
2834                 seq_printf(m, "%.*s\t", OCFS2_LOCK_ID_MAX_LEN, lockres->l_name);
2835 
2836         seq_printf(m, "%d\t"
2837                    "0x%lx\t"
2838                    "0x%x\t"
2839                    "0x%x\t"
2840                    "%u\t"
2841                    "%u\t"
2842                    "%d\t"
2843                    "%d\t",
2844                    lockres->l_level,
2845                    lockres->l_flags,
2846                    lockres->l_action,
2847                    lockres->l_unlock_action,
2848                    lockres->l_ro_holders,
2849                    lockres->l_ex_holders,
2850                    lockres->l_requested,
2851                    lockres->l_blocking);
2852 
2853         /* Dump the raw LVB */
2854         lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
2855         for(i = 0; i < DLM_LVB_LEN; i++)
2856                 seq_printf(m, "0x%x\t", lvb[i]);
2857 
2858 #ifdef CONFIG_OCFS2_FS_STATS
2859 # define lock_num_prmode(_l)            ((_l)->l_lock_prmode.ls_gets)
2860 # define lock_num_exmode(_l)            ((_l)->l_lock_exmode.ls_gets)
2861 # define lock_num_prmode_failed(_l)     ((_l)->l_lock_prmode.ls_fail)
2862 # define lock_num_exmode_failed(_l)     ((_l)->l_lock_exmode.ls_fail)
2863 # define lock_total_prmode(_l)          ((_l)->l_lock_prmode.ls_total)
2864 # define lock_total_exmode(_l)          ((_l)->l_lock_exmode.ls_total)
2865 # define lock_max_prmode(_l)            ((_l)->l_lock_prmode.ls_max)
2866 # define lock_max_exmode(_l)            ((_l)->l_lock_exmode.ls_max)
2867 # define lock_refresh(_l)               ((_l)->l_lock_refresh)
2868 #else
2869 # define lock_num_prmode(_l)            (0)
2870 # define lock_num_exmode(_l)            (0)
2871 # define lock_num_prmode_failed(_l)     (0)
2872 # define lock_num_exmode_failed(_l)     (0)
2873 # define lock_total_prmode(_l)          (0ULL)
2874 # define lock_total_exmode(_l)          (0ULL)
2875 # define lock_max_prmode(_l)            (0)
2876 # define lock_max_exmode(_l)            (0)
2877 # define lock_refresh(_l)               (0)
2878 #endif
2879         /* The following seq_print was added in version 2 of this output */
2880         seq_printf(m, "%u\t"
2881                    "%u\t"
2882                    "%u\t"
2883                    "%u\t"
2884                    "%llu\t"
2885                    "%llu\t"
2886                    "%u\t"
2887                    "%u\t"
2888                    "%u\t",
2889                    lock_num_prmode(lockres),
2890                    lock_num_exmode(lockres),
2891                    lock_num_prmode_failed(lockres),
2892                    lock_num_exmode_failed(lockres),
2893                    lock_total_prmode(lockres),
2894                    lock_total_exmode(lockres),
2895                    lock_max_prmode(lockres),
2896                    lock_max_exmode(lockres),
2897                    lock_refresh(lockres));
2898 
2899         /* End the line */
2900         seq_printf(m, "\n");
2901         return 0;
2902 }
2903 
2904 static const struct seq_operations ocfs2_dlm_seq_ops = {
2905         .start =        ocfs2_dlm_seq_start,
2906         .stop =         ocfs2_dlm_seq_stop,
2907         .next =         ocfs2_dlm_seq_next,
2908         .show =         ocfs2_dlm_seq_show,
2909 };
2910 
2911 static int ocfs2_dlm_debug_release(struct inode *inode, struct file *file)
2912 {
2913         struct seq_file *seq = file->private_data;
2914         struct ocfs2_dlm_seq_priv *priv = seq->private;
2915         struct ocfs2_lock_res *res = &priv->p_iter_res;
2916 
2917         ocfs2_remove_lockres_tracking(res);
2918         ocfs2_put_dlm_debug(priv->p_dlm_debug);
2919         return seq_release_private(inode, file);
2920 }
2921 
2922 static int ocfs2_dlm_debug_open(struct inode *inode, struct file *file)
2923 {
2924         struct ocfs2_dlm_seq_priv *priv;
2925         struct ocfs2_super *osb;
2926 
2927         priv = __seq_open_private(file, &ocfs2_dlm_seq_ops, sizeof(*priv));
2928         if (!priv) {
2929                 mlog_errno(-ENOMEM);
2930                 return -ENOMEM;
2931         }
2932 
2933         osb = inode->i_private;
2934         ocfs2_get_dlm_debug(osb->osb_dlm_debug);
2935         priv->p_dlm_debug = osb->osb_dlm_debug;
2936         INIT_LIST_HEAD(&priv->p_iter_res.l_debug_list);
2937 
2938         ocfs2_add_lockres_tracking(&priv->p_iter_res,
2939                                    priv->p_dlm_debug);
2940 
2941         return 0;
2942 }
2943 
2944 static const struct file_operations ocfs2_dlm_debug_fops = {
2945         .open =         ocfs2_dlm_debug_open,
2946         .release =      ocfs2_dlm_debug_release,
2947         .read =         seq_read,
2948         .llseek =       seq_lseek,
2949 };
2950 
2951 static int ocfs2_dlm_init_debug(struct ocfs2_super *osb)
2952 {
2953         int ret = 0;
2954         struct ocfs2_dlm_debug *dlm_debug = osb->osb_dlm_debug;
2955 
2956         dlm_debug->d_locking_state = debugfs_create_file("locking_state",
2957                                                          S_IFREG|S_IRUSR,
2958                                                          osb->osb_debug_root,
2959                                                          osb,
2960                                                          &ocfs2_dlm_debug_fops);
2961         if (!dlm_debug->d_locking_state) {
2962                 ret = -EINVAL;
2963                 mlog(ML_ERROR,
2964                      "Unable to create locking state debugfs file.\n");
2965                 goto out;
2966         }
2967 
2968         ocfs2_get_dlm_debug(dlm_debug);
2969 out:
2970         return ret;
2971 }
2972 
2973 static void ocfs2_dlm_shutdown_debug(struct ocfs2_super *osb)
2974 {
2975         struct ocfs2_dlm_debug *dlm_debug = osb->osb_dlm_debug;
2976 
2977         if (dlm_debug) {
2978                 debugfs_remove(dlm_debug->d_locking_state);
2979                 ocfs2_put_dlm_debug(dlm_debug);
2980         }
2981 }
2982 
2983 int ocfs2_dlm_init(struct ocfs2_super *osb)
2984 {
2985         int status = 0;
2986         struct ocfs2_cluster_connection *conn = NULL;
2987 
2988         if (ocfs2_mount_local(osb)) {
2989                 osb->node_num = 0;
2990                 goto local;
2991         }
2992 
2993         status = ocfs2_dlm_init_debug(osb);
2994         if (status < 0) {
2995                 mlog_errno(status);
2996                 goto bail;
2997         }
2998 
2999         /* launch downconvert thread */
3000         osb->dc_task = kthread_run(ocfs2_downconvert_thread, osb, "ocfs2dc-%s",
3001                         osb->uuid_str);
3002         if (IS_ERR(osb->dc_task)) {
3003                 status = PTR_ERR(osb->dc_task);
3004                 osb->dc_task = NULL;
3005                 mlog_errno(status);
3006                 goto bail;
3007         }
3008 
3009         /* for now, uuid == domain */
3010         status = ocfs2_cluster_connect(osb->osb_cluster_stack,
3011                                        osb->osb_cluster_name,
3012                                        strlen(osb->osb_cluster_name),
3013                                        osb->uuid_str,
3014                                        strlen(osb->uuid_str),
3015                                        &lproto, ocfs2_do_node_down, osb,
3016                                        &conn);
3017         if (status) {
3018                 mlog_errno(status);
3019                 goto bail;
3020         }
3021 
3022         status = ocfs2_cluster_this_node(conn, &osb->node_num);
3023         if (status < 0) {
3024                 mlog_errno(status);
3025                 mlog(ML_ERROR,
3026                      "could not find this host's node number\n");
3027                 ocfs2_cluster_disconnect(conn, 0);
3028                 goto bail;
3029         }
3030 
3031 local:
3032         ocfs2_super_lock_res_init(&osb->osb_super_lockres, osb);
3033         ocfs2_rename_lock_res_init(&osb->osb_rename_lockres, osb);
3034         ocfs2_nfs_sync_lock_res_init(&osb->osb_nfs_sync_lockres, osb);
3035         ocfs2_orphan_scan_lock_res_init(&osb->osb_orphan_scan.os_lockres, osb);
3036 
3037         osb->cconn = conn;
3038 bail:
3039         if (status < 0) {
3040                 ocfs2_dlm_shutdown_debug(osb);
3041                 if (osb->dc_task)
3042                         kthread_stop(osb->dc_task);
3043         }
3044 
3045         return status;
3046 }
3047 
3048 void ocfs2_dlm_shutdown(struct ocfs2_super *osb,
3049                         int hangup_pending)
3050 {
3051         ocfs2_drop_osb_locks(osb);
3052 
3053         /*
3054          * Now that we have dropped all locks and ocfs2_dismount_volume()
3055          * has disabled recovery, the DLM won't be talking to us.  It's
3056          * safe to tear things down before disconnecting the cluster.
3057          */
3058 
3059         if (osb->dc_task) {
3060                 kthread_stop(osb->dc_task);
3061                 osb->dc_task = NULL;
3062         }
3063 
3064         ocfs2_lock_res_free(&osb->osb_super_lockres);
3065         ocfs2_lock_res_free(&osb->osb_rename_lockres);
3066         ocfs2_lock_res_free(&osb->osb_nfs_sync_lockres);
3067         ocfs2_lock_res_free(&osb->osb_orphan_scan.os_lockres);
3068 
3069         ocfs2_cluster_disconnect(osb->cconn, hangup_pending);
3070         osb->cconn = NULL;
3071 
3072         ocfs2_dlm_shutdown_debug(osb);
3073 }
3074 
3075 static int ocfs2_drop_lock(struct ocfs2_super *osb,
3076                            struct ocfs2_lock_res *lockres)
3077 {
3078         int ret;
3079         unsigned long flags;
3080         u32 lkm_flags = 0;
3081 
3082         /* We didn't get anywhere near actually using this lockres. */
3083         if (!(lockres->l_flags & OCFS2_LOCK_INITIALIZED))
3084                 goto out;
3085 
3086         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
3087                 lkm_flags |= DLM_LKF_VALBLK;
3088 
3089         spin_lock_irqsave(&lockres->l_lock, flags);
3090 
3091         mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_FREEING),
3092                         "lockres %s, flags 0x%lx\n",
3093                         lockres->l_name, lockres->l_flags);
3094 
3095         while (lockres->l_flags & OCFS2_LOCK_BUSY) {
3096                 mlog(0, "waiting on busy lock \"%s\": flags = %lx, action = "
3097                      "%u, unlock_action = %u\n",
3098                      lockres->l_name, lockres->l_flags, lockres->l_action,
3099                      lockres->l_unlock_action);
3100 
3101                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3102 
3103                 /* XXX: Today we just wait on any busy
3104                  * locks... Perhaps we need to cancel converts in the
3105                  * future? */
3106                 ocfs2_wait_on_busy_lock(lockres);
3107 
3108                 spin_lock_irqsave(&lockres->l_lock, flags);
3109         }
3110 
3111         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
3112                 if (lockres->l_flags & OCFS2_LOCK_ATTACHED &&
3113                     lockres->l_level == DLM_LOCK_EX &&
3114                     !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
3115                         lockres->l_ops->set_lvb(lockres);
3116         }
3117 
3118         if (lockres->l_flags & OCFS2_LOCK_BUSY)
3119                 mlog(ML_ERROR, "destroying busy lock: \"%s\"\n",
3120                      lockres->l_name);
3121         if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
3122                 mlog(0, "destroying blocked lock: \"%s\"\n", lockres->l_name);
3123 
3124         if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
3125                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3126                 goto out;
3127         }
3128 
3129         lockres_clear_flags(lockres, OCFS2_LOCK_ATTACHED);
3130 
3131         /* make sure we never get here while waiting for an ast to
3132          * fire. */
3133         BUG_ON(lockres->l_action != OCFS2_AST_INVALID);
3134 
3135         /* is this necessary? */
3136         lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
3137         lockres->l_unlock_action = OCFS2_UNLOCK_DROP_LOCK;
3138         spin_unlock_irqrestore(&lockres->l_lock, flags);
3139 
3140         mlog(0, "lock %s\n", lockres->l_name);
3141 
3142         ret = ocfs2_dlm_unlock(osb->cconn, &lockres->l_lksb, lkm_flags);
3143         if (ret) {
3144                 ocfs2_log_dlm_error("ocfs2_dlm_unlock", ret, lockres);
3145                 mlog(ML_ERROR, "lockres flags: %lu\n", lockres->l_flags);
3146                 ocfs2_dlm_dump_lksb(&lockres->l_lksb);
3147                 BUG();
3148         }
3149         mlog(0, "lock %s, successful return from ocfs2_dlm_unlock\n",
3150              lockres->l_name);
3151 
3152         ocfs2_wait_on_busy_lock(lockres);
3153 out:
3154         return 0;
3155 }
3156 
3157 static void ocfs2_process_blocked_lock(struct ocfs2_super *osb,
3158                                        struct ocfs2_lock_res *lockres);
3159 
3160 /* Mark the lockres as being dropped. It will no longer be
3161  * queued if blocking, but we still may have to wait on it
3162  * being dequeued from the downconvert thread before we can consider
3163  * it safe to drop.
3164  *
3165  * You can *not* attempt to call cluster_lock on this lockres anymore. */
3166 void ocfs2_mark_lockres_freeing(struct ocfs2_super *osb,
3167                                 struct ocfs2_lock_res *lockres)
3168 {
3169         int status;
3170         struct ocfs2_mask_waiter mw;
3171         unsigned long flags, flags2;
3172 
3173         ocfs2_init_mask_waiter(&mw);
3174 
3175         spin_lock_irqsave(&lockres->l_lock, flags);
3176         lockres->l_flags |= OCFS2_LOCK_FREEING;
3177         if (lockres->l_flags & OCFS2_LOCK_QUEUED && current == osb->dc_task) {
3178                 /*
3179                  * We know the downconvert is queued but not in progress
3180                  * because we are the downconvert thread and processing
3181                  * different lock. So we can just remove the lock from the
3182                  * queue. This is not only an optimization but also a way
3183                  * to avoid the following deadlock:
3184                  *   ocfs2_dentry_post_unlock()
3185                  *     ocfs2_dentry_lock_put()
3186                  *       ocfs2_drop_dentry_lock()
3187                  *         iput()
3188                  *           ocfs2_evict_inode()
3189                  *             ocfs2_clear_inode()
3190                  *               ocfs2_mark_lockres_freeing()
3191                  *                 ... blocks waiting for OCFS2_LOCK_QUEUED
3192                  *                 since we are the downconvert thread which
3193                  *                 should clear the flag.
3194                  */
3195                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3196                 spin_lock_irqsave(&osb->dc_task_lock, flags2);
3197                 list_del_init(&lockres->l_blocked_list);
3198                 osb->blocked_lock_count--;
3199                 spin_unlock_irqrestore(&osb->dc_task_lock, flags2);
3200                 /*
3201                  * Warn if we recurse into another post_unlock call.  Strictly
3202                  * speaking it isn't a problem but we need to be careful if
3203                  * that happens (stack overflow, deadlocks, ...) so warn if
3204                  * ocfs2 grows a path for which this can happen.
3205                  */
3206                 WARN_ON_ONCE(lockres->l_ops->post_unlock);
3207                 /* Since the lock is freeing we don't do much in the fn below */
3208                 ocfs2_process_blocked_lock(osb, lockres);
3209                 return;
3210         }
3211         while (lockres->l_flags & OCFS2_LOCK_QUEUED) {
3212                 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_QUEUED, 0);
3213                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3214 
3215                 mlog(0, "Waiting on lockres %s\n", lockres->l_name);
3216 
3217                 status = ocfs2_wait_for_mask(&mw);
3218                 if (status)
3219                         mlog_errno(status);
3220 
3221                 spin_lock_irqsave(&lockres->l_lock, flags);
3222         }
3223         spin_unlock_irqrestore(&lockres->l_lock, flags);
3224 }
3225 
3226 void ocfs2_simple_drop_lockres(struct ocfs2_super *osb,
3227                                struct ocfs2_lock_res *lockres)
3228 {
3229         int ret;
3230 
3231         ocfs2_mark_lockres_freeing(osb, lockres);
3232         ret = ocfs2_drop_lock(osb, lockres);
3233         if (ret)
3234                 mlog_errno(ret);
3235 }
3236 
3237 static void ocfs2_drop_osb_locks(struct ocfs2_super *osb)
3238 {
3239         ocfs2_simple_drop_lockres(osb, &osb->osb_super_lockres);
3240         ocfs2_simple_drop_lockres(osb, &osb->osb_rename_lockres);
3241         ocfs2_simple_drop_lockres(osb, &osb->osb_nfs_sync_lockres);
3242         ocfs2_simple_drop_lockres(osb, &osb->osb_orphan_scan.os_lockres);
3243 }
3244 
3245 int ocfs2_drop_inode_locks(struct inode *inode)
3246 {
3247         int status, err;
3248 
3249         /* No need to call ocfs2_mark_lockres_freeing here -
3250          * ocfs2_clear_inode has done it for us. */
3251 
3252         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3253                               &OCFS2_I(inode)->ip_open_lockres);
3254         if (err < 0)
3255                 mlog_errno(err);
3256 
3257         status = err;
3258 
3259         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3260                               &OCFS2_I(inode)->ip_inode_lockres);
3261         if (err < 0)
3262                 mlog_errno(err);
3263         if (err < 0 && !status)
3264                 status = err;
3265 
3266         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3267                               &OCFS2_I(inode)->ip_rw_lockres);
3268         if (err < 0)
3269                 mlog_errno(err);
3270         if (err < 0 && !status)
3271                 status = err;
3272 
3273         return status;
3274 }
3275 
3276 static unsigned int ocfs2_prepare_downconvert(struct ocfs2_lock_res *lockres,
3277                                               int new_level)
3278 {
3279         assert_spin_locked(&lockres->l_lock);
3280 
3281         BUG_ON(lockres->l_blocking <= DLM_LOCK_NL);
3282 
3283         if (lockres->l_level <= new_level) {
3284                 mlog(ML_ERROR, "lockres %s, lvl %d <= %d, blcklst %d, mask %d, "
3285                      "type %d, flags 0x%lx, hold %d %d, act %d %d, req %d, "
3286                      "block %d, pgen %d\n", lockres->l_name, lockres->l_level,
3287                      new_level, list_empty(&lockres->l_blocked_list),
3288                      list_empty(&lockres->l_mask_waiters), lockres->l_type,
3289                      lockres->l_flags, lockres->l_ro_holders,
3290                      lockres->l_ex_holders, lockres->l_action,
3291                      lockres->l_unlock_action, lockres->l_requested,
3292                      lockres->l_blocking, lockres->l_pending_gen);
3293                 BUG();
3294         }
3295 
3296         mlog(ML_BASTS, "lockres %s, level %d => %d, blocking %d\n",
3297              lockres->l_name, lockres->l_level, new_level, lockres->l_blocking);
3298 
3299         lockres->l_action = OCFS2_AST_DOWNCONVERT;
3300         lockres->l_requested = new_level;
3301         lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
3302         return lockres_set_pending(lockres);
3303 }
3304 
3305 static int ocfs2_downconvert_lock(struct ocfs2_super *osb,
3306                                   struct ocfs2_lock_res *lockres,
3307                                   int new_level,
3308                                   int lvb,
3309                                   unsigned int generation)
3310 {
3311         int ret;
3312         u32 dlm_flags = DLM_LKF_CONVERT;
3313 
3314         mlog(ML_BASTS, "lockres %s, level %d => %d\n", lockres->l_name,
3315              lockres->l_level, new_level);
3316 
3317         if (lvb)
3318                 dlm_flags |= DLM_LKF_VALBLK;
3319 
3320         ret = ocfs2_dlm_lock(osb->cconn,
3321                              new_level,
3322                              &lockres->l_lksb,
3323                              dlm_flags,
3324                              lockres->l_name,
3325                              OCFS2_LOCK_ID_MAX_LEN - 1);
3326         lockres_clear_pending(lockres, generation, osb);
3327         if (ret) {
3328                 ocfs2_log_dlm_error("ocfs2_dlm_lock", ret, lockres);
3329                 ocfs2_recover_from_dlm_error(lockres, 1);
3330                 goto bail;
3331         }
3332 
3333         ret = 0;
3334 bail:
3335         return ret;
3336 }
3337 
3338 /* returns 1 when the caller should unlock and call ocfs2_dlm_unlock */
3339 static int ocfs2_prepare_cancel_convert(struct ocfs2_super *osb,
3340                                         struct ocfs2_lock_res *lockres)
3341 {
3342         assert_spin_locked(&lockres->l_lock);
3343 
3344         if (lockres->l_unlock_action == OCFS2_UNLOCK_CANCEL_CONVERT) {
3345                 /* If we're already trying to cancel a lock conversion
3346                  * then just drop the spinlock and allow the caller to
3347                  * requeue this lock. */
3348                 mlog(ML_BASTS, "lockres %s, skip convert\n", lockres->l_name);
3349                 return 0;
3350         }
3351 
3352         /* were we in a convert when we got the bast fire? */
3353         BUG_ON(lockres->l_action != OCFS2_AST_CONVERT &&
3354                lockres->l_action != OCFS2_AST_DOWNCONVERT);
3355         /* set things up for the unlockast to know to just
3356          * clear out the ast_action and unset busy, etc. */
3357         lockres->l_unlock_action = OCFS2_UNLOCK_CANCEL_CONVERT;
3358 
3359         mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_BUSY),
3360                         "lock %s, invalid flags: 0x%lx\n",
3361                         lockres->l_name, lockres->l_flags);
3362 
3363         mlog(ML_BASTS, "lockres %s\n", lockres->l_name);
3364 
3365         return 1;
3366 }
3367 
3368 static int ocfs2_cancel_convert(struct ocfs2_super *osb,
3369                                 struct ocfs2_lock_res *lockres)
3370 {
3371         int ret;
3372 
3373         ret = ocfs2_dlm_unlock(osb->cconn, &lockres->l_lksb,
3374                                DLM_LKF_CANCEL);
3375         if (ret) {
3376                 ocfs2_log_dlm_error("ocfs2_dlm_unlock", ret, lockres);
3377                 ocfs2_recover_from_dlm_error(lockres, 0);
3378         }
3379 
3380         mlog(ML_BASTS, "lockres %s\n", lockres->l_name);
3381 
3382         return ret;
3383 }
3384 
3385 static int ocfs2_unblock_lock(struct ocfs2_super *osb,
3386                               struct ocfs2_lock_res *lockres,
3387                               struct ocfs2_unblock_ctl *ctl)
3388 {
3389         unsigned long flags;
3390         int blocking;
3391         int new_level;
3392         int level;
3393         int ret = 0;
3394         int set_lvb = 0;
3395         unsigned int gen;
3396 
3397         spin_lock_irqsave(&lockres->l_lock, flags);
3398 
3399 recheck:
3400         /*
3401          * Is it still blocking? If not, we have no more work to do.
3402          */
3403         if (!(lockres->l_flags & OCFS2_LOCK_BLOCKED)) {
3404                 BUG_ON(lockres->l_blocking != DLM_LOCK_NL);
3405                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3406                 ret = 0;
3407                 goto leave;
3408         }
3409 
3410         if (lockres->l_flags & OCFS2_LOCK_BUSY) {
3411                 /* XXX
3412                  * This is a *big* race.  The OCFS2_LOCK_PENDING flag
3413                  * exists entirely for one reason - another thread has set
3414                  * OCFS2_LOCK_BUSY, but has *NOT* yet called dlm_lock().
3415                  *
3416                  * If we do ocfs2_cancel_convert() before the other thread
3417                  * calls dlm_lock(), our cancel will do nothing.  We will
3418                  * get no ast, and we will have no way of knowing the
3419                  * cancel failed.  Meanwhile, the other thread will call
3420                  * into dlm_lock() and wait...forever.
3421                  *
3422                  * Why forever?  Because another node has asked for the
3423                  * lock first; that's why we're here in unblock_lock().
3424                  *
3425                  * The solution is OCFS2_LOCK_PENDING.  When PENDING is
3426                  * set, we just requeue the unblock.  Only when the other
3427                  * thread has called dlm_lock() and cleared PENDING will
3428                  * we then cancel their request.
3429                  *
3430                  * All callers of dlm_lock() must set OCFS2_DLM_PENDING
3431                  * at the same time they set OCFS2_DLM_BUSY.  They must
3432                  * clear OCFS2_DLM_PENDING after dlm_lock() returns.
3433                  */
3434                 if (lockres->l_flags & OCFS2_LOCK_PENDING) {
3435                         mlog(ML_BASTS, "lockres %s, ReQ: Pending\n",
3436                              lockres->l_name);
3437                         goto leave_requeue;
3438                 }
3439 
3440                 ctl->requeue = 1;
3441                 ret = ocfs2_prepare_cancel_convert(osb, lockres);
3442                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3443                 if (ret) {
3444                         ret = ocfs2_cancel_convert(osb, lockres);
3445                         if (ret < 0)
3446                                 mlog_errno(ret);
3447                 }
3448                 goto leave;
3449         }
3450 
3451         /*
3452          * This prevents livelocks. OCFS2_LOCK_UPCONVERT_FINISHING flag is
3453          * set when the ast is received for an upconvert just before the
3454          * OCFS2_LOCK_BUSY flag is cleared. Now if the fs received a bast
3455          * on the heels of the ast, we want to delay the downconvert just
3456          * enough to allow the up requestor to do its task. Because this
3457          * lock is in the blocked queue, the lock will be downconverted
3458          * as soon as the requestor is done with the lock.
3459          */
3460         if (lockres->l_flags & OCFS2_LOCK_UPCONVERT_FINISHING)
3461                 goto leave_requeue;
3462 
3463         /*
3464          * How can we block and yet be at NL?  We were trying to upconvert
3465          * from NL and got canceled.  The code comes back here, and now
3466          * we notice and clear BLOCKING.
3467          */
3468         if (lockres->l_level == DLM_LOCK_NL) {
3469                 BUG_ON(lockres->l_ex_holders || lockres->l_ro_holders);
3470                 mlog(ML_BASTS, "lockres %s, Aborting dc\n", lockres->l_name);
3471                 lockres->l_blocking = DLM_LOCK_NL;
3472                 lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED);
3473                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3474                 goto leave;
3475         }
3476 
3477         /* if we're blocking an exclusive and we have *any* holders,
3478          * then requeue. */
3479         if ((lockres->l_blocking == DLM_LOCK_EX)
3480             && (lockres->l_ex_holders || lockres->l_ro_holders)) {
3481                 mlog(ML_BASTS, "lockres %s, ReQ: EX/PR Holders %u,%u\n",
3482                      lockres->l_name, lockres->l_ex_holders,
3483                      lockres->l_ro_holders);
3484                 goto leave_requeue;
3485         }
3486 
3487         /* If it's a PR we're blocking, then only
3488          * requeue if we've got any EX holders */
3489         if (lockres->l_blocking == DLM_LOCK_PR &&
3490             lockres->l_ex_holders) {
3491                 mlog(ML_BASTS, "lockres %s, ReQ: EX Holders %u\n",
3492                      lockres->l_name, lockres->l_ex_holders);
3493                 goto leave_requeue;
3494         }
3495 
3496         /*
3497          * Can we get a lock in this state if the holder counts are
3498          * zero? The meta data unblock code used to check this.
3499          */
3500         if ((lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
3501             && (lockres->l_flags & OCFS2_LOCK_REFRESHING)) {
3502                 mlog(ML_BASTS, "lockres %s, ReQ: Lock Refreshing\n",
3503                      lockres->l_name);
3504                 goto leave_requeue;
3505         }
3506 
3507         new_level = ocfs2_highest_compat_lock_level(lockres->l_blocking);
3508 
3509         if (lockres->l_ops->check_downconvert
3510             && !lockres->l_ops->check_downconvert(lockres, new_level)) {
3511                 mlog(ML_BASTS, "lockres %s, ReQ: Checkpointing\n",
3512                      lockres->l_name);
3513                 goto leave_requeue;
3514         }
3515 
3516         /* If we get here, then we know that there are no more
3517          * incompatible holders (and anyone asking for an incompatible
3518          * lock is blocked). We can now downconvert the lock */
3519         if (!lockres->l_ops->downconvert_worker)
3520                 goto downconvert;
3521 
3522         /* Some lockres types want to do a bit of work before
3523          * downconverting a lock. Allow that here. The worker function
3524          * may sleep, so we save off a copy of what we're blocking as
3525          * it may change while we're not holding the spin lock. */
3526         blocking = lockres->l_blocking;
3527         level = lockres->l_level;
3528         spin_unlock_irqrestore(&lockres->l_lock, flags);
3529 
3530         ctl->unblock_action = lockres->l_ops->downconvert_worker(lockres, blocking);
3531 
3532         if (ctl->unblock_action == UNBLOCK_STOP_POST) {
3533                 mlog(ML_BASTS, "lockres %s, UNBLOCK_STOP_POST\n",
3534                      lockres->l_name);
3535                 goto leave;
3536         }
3537 
3538         spin_lock_irqsave(&lockres->l_lock, flags);
3539         if ((blocking != lockres->l_blocking) || (level != lockres->l_level)) {
3540                 /* If this changed underneath us, then we can't drop
3541                  * it just yet. */
3542                 mlog(ML_BASTS, "lockres %s, block=%d:%d, level=%d:%d, "
3543                      "Recheck\n", lockres->l_name, blocking,
3544                      lockres->l_blocking, level, lockres->l_level);
3545                 goto recheck;
3546         }
3547 
3548 downconvert:
3549         ctl->requeue = 0;
3550 
3551         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
3552                 if (lockres->l_level == DLM_LOCK_EX)
3553                         set_lvb = 1;
3554 
3555                 /*
3556                  * We only set the lvb if the lock has been fully
3557                  * refreshed - otherwise we risk setting stale
3558                  * data. Otherwise, there's no need to actually clear
3559                  * out the lvb here as it's value is still valid.
3560                  */
3561                 if (set_lvb && !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
3562                         lockres->l_ops->set_lvb(lockres);
3563         }
3564 
3565         gen = ocfs2_prepare_downconvert(lockres, new_level);
3566         spin_unlock_irqrestore(&lockres->l_lock, flags);
3567         ret = ocfs2_downconvert_lock(osb, lockres, new_level, set_lvb,
3568                                      gen);
3569 
3570 leave:
3571         if (ret)
3572                 mlog_errno(ret);
3573         return ret;
3574 
3575 leave_requeue:
3576         spin_unlock_irqrestore(&lockres->l_lock, flags);
3577         ctl->requeue = 1;
3578 
3579         return 0;
3580 }
3581 
3582 static int ocfs2_data_convert_worker(struct ocfs2_lock_res *lockres,
3583                                      int blocking)
3584 {
3585         struct inode *inode;
3586         struct address_space *mapping;
3587         struct ocfs2_inode_info *oi;
3588 
3589         inode = ocfs2_lock_res_inode(lockres);
3590         mapping = inode->i_mapping;
3591 
3592         if (S_ISDIR(inode->i_mode)) {
3593                 oi = OCFS2_I(inode);
3594                 oi->ip_dir_lock_gen++;
3595                 mlog(0, "generation: %u\n", oi->ip_dir_lock_gen);
3596                 goto out;
3597         }
3598 
3599         if (!S_ISREG(inode->i_mode))
3600                 goto out;
3601 
3602         /*
3603          * We need this before the filemap_fdatawrite() so that it can
3604          * transfer the dirty bit from the PTE to the
3605          * page. Unfortunately this means that even for EX->PR
3606          * downconverts, we'll lose our mappings and have to build
3607          * them up again.
3608          */
3609         unmap_mapping_range(mapping, 0, 0, 0);
3610 
3611         if (filemap_fdatawrite(mapping)) {
3612                 mlog(ML_ERROR, "Could not sync inode %llu for downconvert!",
3613                      (unsigned long long)OCFS2_I(inode)->ip_blkno);
3614         }
3615         sync_mapping_buffers(mapping);
3616         if (blocking == DLM_LOCK_EX) {
3617                 truncate_inode_pages(mapping, 0);
3618         } else {
3619                 /* We only need to wait on the I/O if we're not also
3620                  * truncating pages because truncate_inode_pages waits
3621                  * for us above. We don't truncate pages if we're
3622                  * blocking anything < EXMODE because we want to keep
3623                  * them around in that case. */
3624                 filemap_fdatawait(mapping);
3625         }
3626 
3627         forget_all_cached_acls(inode);
3628 
3629 out:
3630         return UNBLOCK_CONTINUE;
3631 }
3632 
3633 static int ocfs2_ci_checkpointed(struct ocfs2_caching_info *ci,
3634                                  struct ocfs2_lock_res *lockres,
3635                                  int new_level)
3636 {
3637         int checkpointed = ocfs2_ci_fully_checkpointed(ci);
3638 
3639         BUG_ON(new_level != DLM_LOCK_NL && new_level != DLM_LOCK_PR);
3640         BUG_ON(lockres->l_level != DLM_LOCK_EX && !checkpointed);
3641 
3642         if (checkpointed)
3643                 return 1;
3644 
3645         ocfs2_start_checkpoint(OCFS2_SB(ocfs2_metadata_cache_get_super(ci)));
3646         return 0;
3647 }
3648 
3649 static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
3650                                         int new_level)
3651 {
3652         struct inode *inode = ocfs2_lock_res_inode(lockres);
3653 
3654         return ocfs2_ci_checkpointed(INODE_CACHE(inode), lockres, new_level);
3655 }
3656 
3657 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres)
3658 {
3659         struct inode *inode = ocfs2_lock_res_inode(lockres);
3660 
3661         __ocfs2_stuff_meta_lvb(inode);
3662 }
3663 
3664 /*
3665  * Does the final reference drop on our dentry lock. Right now this
3666  * happens in the downconvert thread, but we could choose to simplify the
3667  * dlmglue API and push these off to the ocfs2_wq in the future.
3668  */
3669 static void ocfs2_dentry_post_unlock(struct ocfs2_super *osb,
3670                                      struct ocfs2_lock_res *lockres)
3671 {
3672         struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
3673         ocfs2_dentry_lock_put(osb, dl);
3674 }
3675 
3676 /*
3677  * d_delete() matching dentries before the lock downconvert.
3678  *
3679  * At this point, any process waiting to destroy the
3680  * dentry_lock due to last ref count is stopped by the
3681  * OCFS2_LOCK_QUEUED flag.
3682  *
3683  * We have two potential problems
3684  *
3685  * 1) If we do the last reference drop on our dentry_lock (via dput)
3686  *    we'll wind up in ocfs2_release_dentry_lock(), waiting on
3687  *    the downconvert to finish. Instead we take an elevated
3688  *    reference and push the drop until after we've completed our
3689  *    unblock processing.
3690  *
3691  * 2) There might be another process with a final reference,
3692  *    waiting on us to finish processing. If this is the case, we
3693  *    detect it and exit out - there's no more dentries anyway.
3694  */
3695 static int ocfs2_dentry_convert_worker(struct ocfs2_lock_res *lockres,
3696                                        int blocking)
3697 {
3698         struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
3699         struct ocfs2_inode_info *oi = OCFS2_I(dl->dl_inode);
3700         struct dentry *dentry;
3701         unsigned long flags;
3702         int extra_ref = 0;
3703 
3704         /*
3705          * This node is blocking another node from getting a read
3706          * lock. This happens when we've renamed within a
3707          * directory. We've forced the other nodes to d_delete(), but
3708          * we never actually dropped our lock because it's still
3709          * valid. The downconvert code will retain a PR for this node,
3710          * so there's no further work to do.
3711          */
3712         if (blocking == DLM_LOCK_PR)
3713                 return UNBLOCK_CONTINUE;
3714 
3715         /*
3716          * Mark this inode as potentially orphaned. The code in
3717          * ocfs2_delete_inode() will figure out whether it actually
3718          * needs to be freed or not.
3719          */
3720         spin_lock(&oi->ip_lock);
3721         oi->ip_flags |= OCFS2_INODE_MAYBE_ORPHANED;
3722         spin_unlock(&oi->ip_lock);
3723 
3724         /*
3725          * Yuck. We need to make sure however that the check of
3726          * OCFS2_LOCK_FREEING and the extra reference are atomic with
3727          * respect to a reference decrement or the setting of that
3728          * flag.
3729          */
3730         spin_lock_irqsave(&lockres->l_lock, flags);
3731         spin_lock(&dentry_attach_lock);
3732         if (!(lockres->l_flags & OCFS2_LOCK_FREEING)
3733             && dl->dl_count) {
3734                 dl->dl_count++;
3735                 extra_ref = 1;
3736         }
3737         spin_unlock(&dentry_attach_lock);
3738         spin_unlock_irqrestore(&lockres->l_lock, flags);
3739 
3740         mlog(0, "extra_ref = %d\n", extra_ref);
3741 
3742         /*
3743          * We have a process waiting on us in ocfs2_dentry_iput(),
3744          * which means we can't have any more outstanding
3745          * aliases. There's no need to do any more work.
3746          */
3747         if (!extra_ref)
3748                 return UNBLOCK_CONTINUE;
3749 
3750         spin_lock(&dentry_attach_lock);
3751         while (1) {
3752                 dentry = ocfs2_find_local_alias(dl->dl_inode,
3753                                                 dl->dl_parent_blkno, 1);
3754                 if (!dentry)
3755                         break;
3756                 spin_unlock(&dentry_attach_lock);
3757 
3758                 if (S_ISDIR(dl->dl_inode->i_mode))
3759                         shrink_dcache_parent(dentry);
3760 
3761                 mlog(0, "d_delete(%pd);\n", dentry);
3762 
3763                 /*
3764                  * The following dcache calls may do an
3765                  * iput(). Normally we don't want that from the
3766                  * downconverting thread, but in this case it's ok
3767                  * because the requesting node already has an
3768                  * exclusive lock on the inode, so it can't be queued
3769                  * for a downconvert.
3770                  */
3771                 d_delete(dentry);
3772                 dput(dentry);
3773 
3774                 spin_lock(&dentry_attach_lock);
3775         }
3776         spin_unlock(&dentry_attach_lock);
3777 
3778         /*
3779          * If we are the last holder of this dentry lock, there is no
3780          * reason to downconvert so skip straight to the unlock.
3781          */
3782         if (dl->dl_count == 1)
3783                 return UNBLOCK_STOP_POST;
3784 
3785         return UNBLOCK_CONTINUE_POST;
3786 }
3787 
3788 static int ocfs2_check_refcount_downconvert(struct ocfs2_lock_res *lockres,
3789                                             int new_level)
3790 {
3791         struct ocfs2_refcount_tree *tree =
3792                                 ocfs2_lock_res_refcount_tree(lockres);
3793 
3794         return ocfs2_ci_checkpointed(&tree->rf_ci, lockres, new_level);
3795 }
3796 
3797 static int ocfs2_refcount_convert_worker(struct ocfs2_lock_res *lockres,
3798                                          int blocking)
3799 {
3800         struct ocfs2_refcount_tree *tree =
3801                                 ocfs2_lock_res_refcount_tree(lockres);
3802 
3803         ocfs2_metadata_cache_purge(&tree->rf_ci);
3804 
3805         return UNBLOCK_CONTINUE;
3806 }
3807 
3808 static void ocfs2_set_qinfo_lvb(struct ocfs2_lock_res *lockres)
3809 {
3810         struct ocfs2_qinfo_lvb *lvb;
3811         struct ocfs2_mem_dqinfo *oinfo = ocfs2_lock_res_qinfo(lockres);
3812         struct mem_dqinfo *info = sb_dqinfo(oinfo->dqi_gi.dqi_sb,
3813                                             oinfo->dqi_gi.dqi_type);
3814 
3815         lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
3816         lvb->lvb_version = OCFS2_QINFO_LVB_VERSION;
3817         lvb->lvb_bgrace = cpu_to_be32(info->dqi_bgrace);
3818         lvb->lvb_igrace = cpu_to_be32(info->dqi_igrace);
3819         lvb->lvb_syncms = cpu_to_be32(oinfo->dqi_syncms);
3820         lvb->lvb_blocks = cpu_to_be32(oinfo->dqi_gi.dqi_blocks);
3821         lvb->lvb_free_blk = cpu_to_be32(oinfo->dqi_gi.dqi_free_blk);
3822         lvb->lvb_free_entry = cpu_to_be32(oinfo->dqi_gi.dqi_free_entry);
3823 }
3824 
3825 void ocfs2_qinfo_unlock(struct ocfs2_mem_dqinfo *oinfo, int ex)
3826 {
3827         struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
3828         struct ocfs2_super *osb = OCFS2_SB(oinfo->dqi_gi.dqi_sb);
3829         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
3830 
3831         if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb))
3832                 ocfs2_cluster_unlock(osb, lockres, level);
3833 }
3834 
3835 static int ocfs2_refresh_qinfo(struct ocfs2_mem_dqinfo *oinfo)
3836 {
3837         struct mem_dqinfo *info = sb_dqinfo(oinfo->dqi_gi.dqi_sb,
3838                                             oinfo->dqi_gi.dqi_type);
3839         struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
3840         struct ocfs2_qinfo_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
3841         struct buffer_head *bh = NULL;
3842         struct ocfs2_global_disk_dqinfo *gdinfo;
3843         int status = 0;
3844 
3845         if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
3846             lvb->lvb_version == OCFS2_QINFO_LVB_VERSION) {
3847                 info->dqi_bgrace = be32_to_cpu(lvb->lvb_bgrace);
3848                 info->dqi_igrace = be32_to_cpu(lvb->lvb_igrace);
3849                 oinfo->dqi_syncms = be32_to_cpu(lvb->lvb_syncms);
3850                 oinfo->dqi_gi.dqi_blocks = be32_to_cpu(lvb->lvb_blocks);
3851                 oinfo->dqi_gi.dqi_free_blk = be32_to_cpu(lvb->lvb_free_blk);
3852                 oinfo->dqi_gi.dqi_free_entry =
3853                                         be32_to_cpu(lvb->lvb_free_entry);
3854         } else {
3855                 status = ocfs2_read_quota_phys_block(oinfo->dqi_gqinode,
3856                                                      oinfo->dqi_giblk, &bh);
3857                 if (status) {
3858                         mlog_errno(status);
3859                         goto bail;
3860                 }
3861                 gdinfo = (struct ocfs2_global_disk_dqinfo *)
3862                                         (bh->b_data + OCFS2_GLOBAL_INFO_OFF);
3863                 info->dqi_bgrace = le32_to_cpu(gdinfo->dqi_bgrace);
3864                 info->dqi_igrace = le32_to_cpu(gdinfo->dqi_igrace);
3865                 oinfo->dqi_syncms = le32_to_cpu(gdinfo->dqi_syncms);
3866                 oinfo->dqi_gi.dqi_blocks = le32_to_cpu(gdinfo->dqi_blocks);
3867                 oinfo->dqi_gi.dqi_free_blk = le32_to_cpu(gdinfo->dqi_free_blk);
3868                 oinfo->dqi_gi.dqi_free_entry =
3869                                         le32_to_cpu(gdinfo->dqi_free_entry);
3870                 brelse(bh);
3871                 ocfs2_track_lock_refresh(lockres);
3872         }
3873 
3874 bail:
3875         return status;
3876 }
3877 
3878 /* Lock quota info, this function expects at least shared lock on the quota file
3879  * so that we can safely refresh quota info from disk. */
3880 int ocfs2_qinfo_lock(struct ocfs2_mem_dqinfo *oinfo, int ex)
3881 {
3882         struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
3883         struct ocfs2_super *osb = OCFS2_SB(oinfo->dqi_gi.dqi_sb);
3884         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
3885         int status = 0;
3886 
3887         /* On RO devices, locking really isn't needed... */
3888         if (ocfs2_is_hard_readonly(osb)) {
3889                 if (ex)
3890                         status = -EROFS;
3891                 goto bail;
3892         }
3893         if (ocfs2_mount_local(osb))
3894                 goto bail;
3895 
3896         status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
3897         if (status < 0) {
3898                 mlog_errno(status);
3899                 goto bail;
3900         }
3901         if (!ocfs2_should_refresh_lock_res(lockres))
3902                 goto bail;
3903         /* OK, we have the lock but we need to refresh the quota info */
3904         status = ocfs2_refresh_qinfo(oinfo);
3905         if (status)
3906                 ocfs2_qinfo_unlock(oinfo, ex);
3907         ocfs2_complete_lock_res_refresh(lockres, status);
3908 bail:
3909         return status;
3910 }
3911 
3912 int ocfs2_refcount_lock(struct ocfs2_refcount_tree *ref_tree, int ex)
3913 {
3914         int status;
3915         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
3916         struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
3917         struct ocfs2_super *osb = lockres->l_priv;
3918 
3919 
3920         if (ocfs2_is_hard_readonly(osb))
3921                 return -EROFS;
3922 
3923         if (ocfs2_mount_local(osb))
3924                 return 0;
3925 
3926         status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
3927         if (status < 0)
3928                 mlog_errno(status);
3929 
3930         return status;
3931 }
3932 
3933 void ocfs2_refcount_unlock(struct ocfs2_refcount_tree *ref_tree, int ex)
3934 {
3935         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
3936         struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
3937         struct ocfs2_super *osb = lockres->l_priv;
3938 
3939         if (!ocfs2_mount_local(osb))
3940                 ocfs2_cluster_unlock(osb, lockres, level);
3941 }
3942 
3943 static void ocfs2_process_blocked_lock(struct ocfs2_super *osb,
3944                                        struct ocfs2_lock_res *lockres)
3945 {
3946         int status;
3947         struct ocfs2_unblock_ctl ctl = {0, 0,};
3948         unsigned long flags;
3949 
3950         /* Our reference to the lockres in this function can be
3951          * considered valid until we remove the OCFS2_LOCK_QUEUED
3952          * flag. */
3953 
3954         BUG_ON(!lockres);
3955         BUG_ON(!lockres->l_ops);
3956 
3957         mlog(ML_BASTS, "lockres %s blocked\n", lockres->l_name);
3958 
3959         /* Detect whether a lock has been marked as going away while
3960          * the downconvert thread was processing other things. A lock can
3961          * still be marked with OCFS2_LOCK_FREEING after this check,
3962          * but short circuiting here will still save us some
3963          * performance. */
3964         spin_lock_irqsave(&lockres->l_lock, flags);
3965         if (lockres->l_flags & OCFS2_LOCK_FREEING)
3966                 goto unqueue;
3967         spin_unlock_irqrestore(&lockres->l_lock, flags);
3968 
3969         status = ocfs2_unblock_lock(osb, lockres, &ctl);
3970         if (status < 0)
3971                 mlog_errno(status);
3972 
3973         spin_lock_irqsave(&lockres->l_lock, flags);
3974 unqueue:
3975         if (lockres->l_flags & OCFS2_LOCK_FREEING || !ctl.requeue) {
3976                 lockres_clear_flags(lockres, OCFS2_LOCK_QUEUED);
3977         } else
3978                 ocfs2_schedule_blocked_lock(osb, lockres);
3979 
3980         mlog(ML_BASTS, "lockres %s, requeue = %s.\n", lockres->l_name,
3981              ctl.requeue ? "yes" : "no");
3982         spin_unlock_irqrestore(&lockres->l_lock, flags);
3983 
3984         if (ctl.unblock_action != UNBLOCK_CONTINUE
3985             && lockres->l_ops->post_unlock)
3986                 lockres->l_ops->post_unlock(osb, lockres);
3987 }
3988 
3989 static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
3990                                         struct ocfs2_lock_res *lockres)
3991 {
3992         unsigned long flags;
3993 
3994         assert_spin_locked(&lockres->l_lock);
3995 
3996         if (lockres->l_flags & OCFS2_LOCK_FREEING) {
3997                 /* Do not schedule a lock for downconvert when it's on
3998                  * the way to destruction - any nodes wanting access
3999                  * to the resource will get it soon. */
4000                 mlog(ML_BASTS, "lockres %s won't be scheduled: flags 0x%lx\n",
4001                      lockres->l_name, lockres->l_flags);
4002                 return;
4003         }
4004 
4005         lockres_or_flags(lockres, OCFS2_LOCK_QUEUED);
4006 
4007         spin_lock_irqsave(&osb->dc_task_lock, flags);
4008         if (list_empty(&lockres->l_blocked_list)) {
4009                 list_add_tail(&lockres->l_blocked_list,
4010                               &osb->blocked_lock_list);
4011                 osb->blocked_lock_count++;
4012         }
4013         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4014 }
4015 
4016 static void ocfs2_downconvert_thread_do_work(struct ocfs2_super *osb)
4017 {
4018         unsigned long processed;
4019         unsigned long flags;
4020         struct ocfs2_lock_res *lockres;
4021 
4022         spin_lock_irqsave(&osb->dc_task_lock, flags);
4023         /* grab this early so we know to try again if a state change and
4024          * wake happens part-way through our work  */
4025         osb->dc_work_sequence = osb->dc_wake_sequence;
4026 
4027         processed = osb->blocked_lock_count;
4028         /*
4029          * blocked lock processing in this loop might call iput which can
4030          * remove items off osb->blocked_lock_list. Downconvert up to
4031          * 'processed' number of locks, but stop short if we had some
4032          * removed in ocfs2_mark_lockres_freeing when downconverting.
4033          */
4034         while (processed && !list_empty(&osb->blocked_lock_list)) {
4035                 lockres = list_entry(osb->blocked_lock_list.next,
4036                                      struct ocfs2_lock_res, l_blocked_list);
4037                 list_del_init(&lockres->l_blocked_list);
4038                 osb->blocked_lock_count--;
4039                 spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4040 
4041                 BUG_ON(!processed);
4042                 processed--;
4043 
4044                 ocfs2_process_blocked_lock(osb, lockres);
4045 
4046                 spin_lock_irqsave(&osb->dc_task_lock, flags);
4047         }
4048         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4049 }
4050 
4051 static int ocfs2_downconvert_thread_lists_empty(struct ocfs2_super *osb)
4052 {
4053         int empty = 0;
4054         unsigned long flags;
4055 
4056         spin_lock_irqsave(&osb->dc_task_lock, flags);
4057         if (list_empty(&osb->blocked_lock_list))
4058                 empty = 1;
4059 
4060         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4061         return empty;
4062 }
4063 
4064 static int ocfs2_downconvert_thread_should_wake(struct ocfs2_super *osb)
4065 {
4066         int should_wake = 0;
4067         unsigned long flags;
4068 
4069         spin_lock_irqsave(&osb->dc_task_lock, flags);
4070         if (osb->dc_work_sequence != osb->dc_wake_sequence)
4071                 should_wake = 1;
4072         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4073 
4074         return should_wake;
4075 }
4076 
4077 static int ocfs2_downconvert_thread(void *arg)
4078 {
4079         int status = 0;
4080         struct ocfs2_super *osb = arg;
4081 
4082         /* only quit once we've been asked to stop and there is no more
4083          * work available */
4084         while (!(kthread_should_stop() &&
4085                 ocfs2_downconvert_thread_lists_empty(osb))) {
4086 
4087                 wait_event_interruptible(osb->dc_event,
4088                                          ocfs2_downconvert_thread_should_wake(osb) ||
4089                                          kthread_should_stop());
4090 
4091                 mlog(0, "downconvert_thread: awoken\n");
4092 
4093                 ocfs2_downconvert_thread_do_work(osb);
4094         }
4095 
4096         osb->dc_task = NULL;
4097         return status;
4098 }
4099 
4100 void ocfs2_wake_downconvert_thread(struct ocfs2_super *osb)
4101 {
4102         unsigned long flags;
4103 
4104         spin_lock_irqsave(&osb->dc_task_lock, flags);
4105         /* make sure the voting thread gets a swipe at whatever changes
4106          * the caller may have made to the voting state */
4107         osb->dc_wake_sequence++;
4108         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4109         wake_up(&osb->dc_event);
4110 }
4111 

~ [ 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