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

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

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

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