Merge branch 'for-linus' of git://git.kernel.org/pub/scm/linux/kernel/git/shaggy...
[linux-2.6] / fs / ocfs2 / dlmglue.c
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/crc32.h>
31 #include <linux/kthread.h>
32 #include <linux/pagemap.h>
33 #include <linux/debugfs.h>
34 #include <linux/seq_file.h>
35
36 #include <cluster/heartbeat.h>
37 #include <cluster/nodemanager.h>
38 #include <cluster/tcp.h>
39
40 #include <dlm/dlmapi.h>
41
42 #define MLOG_MASK_PREFIX ML_DLM_GLUE
43 #include <cluster/masklog.h>
44
45 #include "ocfs2.h"
46
47 #include "alloc.h"
48 #include "dcache.h"
49 #include "dlmglue.h"
50 #include "extent_map.h"
51 #include "file.h"
52 #include "heartbeat.h"
53 #include "inode.h"
54 #include "journal.h"
55 #include "slot_map.h"
56 #include "super.h"
57 #include "uptodate.h"
58 #include "vote.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 };
69
70 static struct ocfs2_super *ocfs2_get_dentry_osb(struct ocfs2_lock_res *lockres);
71 static struct ocfs2_super *ocfs2_get_inode_osb(struct ocfs2_lock_res *lockres);
72
73 /*
74  * Return value from ->downconvert_worker functions.
75  *
76  * These control the precise actions of ocfs2_unblock_lock()
77  * and ocfs2_process_blocked_lock()
78  *
79  */
80 enum ocfs2_unblock_action {
81         UNBLOCK_CONTINUE        = 0, /* Continue downconvert */
82         UNBLOCK_CONTINUE_POST   = 1, /* Continue downconvert, fire
83                                       * ->post_unlock callback */
84         UNBLOCK_STOP_POST       = 2, /* Do not downconvert, fire
85                                       * ->post_unlock() callback. */
86 };
87
88 struct ocfs2_unblock_ctl {
89         int requeue;
90         enum ocfs2_unblock_action unblock_action;
91 };
92
93 static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
94                                         int new_level);
95 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres);
96
97 static int ocfs2_data_convert_worker(struct ocfs2_lock_res *lockres,
98                                      int blocking);
99
100 static int ocfs2_dentry_convert_worker(struct ocfs2_lock_res *lockres,
101                                        int blocking);
102
103 static void ocfs2_dentry_post_unlock(struct ocfs2_super *osb,
104                                      struct ocfs2_lock_res *lockres);
105
106
107 #define mlog_meta_lvb(__level, __lockres) ocfs2_dump_meta_lvb_info(__level, __PRETTY_FUNCTION__, __LINE__, __lockres)
108
109 /* This aids in debugging situations where a bad LVB might be involved. */
110 static void ocfs2_dump_meta_lvb_info(u64 level,
111                                      const char *function,
112                                      unsigned int line,
113                                      struct ocfs2_lock_res *lockres)
114 {
115         struct ocfs2_meta_lvb *lvb = (struct ocfs2_meta_lvb *) lockres->l_lksb.lvb;
116
117         mlog(level, "LVB information for %s (called from %s:%u):\n",
118              lockres->l_name, function, line);
119         mlog(level, "version: %u, clusters: %u, generation: 0x%x\n",
120              lvb->lvb_version, be32_to_cpu(lvb->lvb_iclusters),
121              be32_to_cpu(lvb->lvb_igeneration));
122         mlog(level, "size: %llu, uid %u, gid %u, mode 0x%x\n",
123              (unsigned long long)be64_to_cpu(lvb->lvb_isize),
124              be32_to_cpu(lvb->lvb_iuid), be32_to_cpu(lvb->lvb_igid),
125              be16_to_cpu(lvb->lvb_imode));
126         mlog(level, "nlink %u, atime_packed 0x%llx, ctime_packed 0x%llx, "
127              "mtime_packed 0x%llx iattr 0x%x\n", be16_to_cpu(lvb->lvb_inlink),
128              (long long)be64_to_cpu(lvb->lvb_iatime_packed),
129              (long long)be64_to_cpu(lvb->lvb_ictime_packed),
130              (long long)be64_to_cpu(lvb->lvb_imtime_packed),
131              be32_to_cpu(lvb->lvb_iattr));
132 }
133
134
135 /*
136  * OCFS2 Lock Resource Operations
137  *
138  * These fine tune the behavior of the generic dlmglue locking infrastructure.
139  *
140  * The most basic of lock types can point ->l_priv to their respective
141  * struct ocfs2_super and allow the default actions to manage things.
142  *
143  * Right now, each lock type also needs to implement an init function,
144  * and trivial lock/unlock wrappers. ocfs2_simple_drop_lockres()
145  * should be called when the lock is no longer needed (i.e., object
146  * destruction time).
147  */
148 struct ocfs2_lock_res_ops {
149         /*
150          * Translate an ocfs2_lock_res * into an ocfs2_super *. Define
151          * this callback if ->l_priv is not an ocfs2_super pointer
152          */
153         struct ocfs2_super * (*get_osb)(struct ocfs2_lock_res *);
154
155         /*
156          * Optionally called in the downconvert (or "vote") thread
157          * after a successful downconvert. The lockres will not be
158          * referenced after this callback is called, so it is safe to
159          * free memory, etc.
160          *
161          * The exact semantics of when this is called are controlled
162          * by ->downconvert_worker()
163          */
164         void (*post_unlock)(struct ocfs2_super *, struct ocfs2_lock_res *);
165
166         /*
167          * Allow a lock type to add checks to determine whether it is
168          * safe to downconvert a lock. Return 0 to re-queue the
169          * downconvert at a later time, nonzero to continue.
170          *
171          * For most locks, the default checks that there are no
172          * incompatible holders are sufficient.
173          *
174          * Called with the lockres spinlock held.
175          */
176         int (*check_downconvert)(struct ocfs2_lock_res *, int);
177
178         /*
179          * Allows a lock type to populate the lock value block. This
180          * is called on downconvert, and when we drop a lock.
181          *
182          * Locks that want to use this should set LOCK_TYPE_USES_LVB
183          * in the flags field.
184          *
185          * Called with the lockres spinlock held.
186          */
187         void (*set_lvb)(struct ocfs2_lock_res *);
188
189         /*
190          * Called from the downconvert thread when it is determined
191          * that a lock will be downconverted. This is called without
192          * any locks held so the function can do work that might
193          * schedule (syncing out data, etc).
194          *
195          * This should return any one of the ocfs2_unblock_action
196          * values, depending on what it wants the thread to do.
197          */
198         int (*downconvert_worker)(struct ocfs2_lock_res *, int);
199
200         /*
201          * LOCK_TYPE_* flags which describe the specific requirements
202          * of a lock type. Descriptions of each individual flag follow.
203          */
204         int flags;
205 };
206
207 /*
208  * Some locks want to "refresh" potentially stale data when a
209  * meaningful (PRMODE or EXMODE) lock level is first obtained. If this
210  * flag is set, the OCFS2_LOCK_NEEDS_REFRESH flag will be set on the
211  * individual lockres l_flags member from the ast function. It is
212  * expected that the locking wrapper will clear the
213  * OCFS2_LOCK_NEEDS_REFRESH flag when done.
214  */
215 #define LOCK_TYPE_REQUIRES_REFRESH 0x1
216
217 /*
218  * Indicate that a lock type makes use of the lock value block. The
219  * ->set_lvb lock type callback must be defined.
220  */
221 #define LOCK_TYPE_USES_LVB              0x2
222
223 static struct ocfs2_lock_res_ops ocfs2_inode_rw_lops = {
224         .get_osb        = ocfs2_get_inode_osb,
225         .flags          = 0,
226 };
227
228 static struct ocfs2_lock_res_ops ocfs2_inode_meta_lops = {
229         .get_osb        = ocfs2_get_inode_osb,
230         .check_downconvert = ocfs2_check_meta_downconvert,
231         .set_lvb        = ocfs2_set_meta_lvb,
232         .flags          = LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_USES_LVB,
233 };
234
235 static struct ocfs2_lock_res_ops ocfs2_inode_data_lops = {
236         .get_osb        = ocfs2_get_inode_osb,
237         .downconvert_worker = ocfs2_data_convert_worker,
238         .flags          = 0,
239 };
240
241 static struct ocfs2_lock_res_ops ocfs2_super_lops = {
242         .flags          = LOCK_TYPE_REQUIRES_REFRESH,
243 };
244
245 static struct ocfs2_lock_res_ops ocfs2_rename_lops = {
246         .flags          = 0,
247 };
248
249 static struct ocfs2_lock_res_ops ocfs2_dentry_lops = {
250         .get_osb        = ocfs2_get_dentry_osb,
251         .post_unlock    = ocfs2_dentry_post_unlock,
252         .downconvert_worker = ocfs2_dentry_convert_worker,
253         .flags          = 0,
254 };
255
256 static struct ocfs2_lock_res_ops ocfs2_inode_open_lops = {
257         .get_osb        = ocfs2_get_inode_osb,
258         .flags          = 0,
259 };
260
261 static inline int ocfs2_is_inode_lock(struct ocfs2_lock_res *lockres)
262 {
263         return lockres->l_type == OCFS2_LOCK_TYPE_META ||
264                 lockres->l_type == OCFS2_LOCK_TYPE_DATA ||
265                 lockres->l_type == OCFS2_LOCK_TYPE_RW ||
266                 lockres->l_type == OCFS2_LOCK_TYPE_OPEN;
267 }
268
269 static inline struct inode *ocfs2_lock_res_inode(struct ocfs2_lock_res *lockres)
270 {
271         BUG_ON(!ocfs2_is_inode_lock(lockres));
272
273         return (struct inode *) lockres->l_priv;
274 }
275
276 static inline struct ocfs2_dentry_lock *ocfs2_lock_res_dl(struct ocfs2_lock_res *lockres)
277 {
278         BUG_ON(lockres->l_type != OCFS2_LOCK_TYPE_DENTRY);
279
280         return (struct ocfs2_dentry_lock *)lockres->l_priv;
281 }
282
283 static inline struct ocfs2_super *ocfs2_get_lockres_osb(struct ocfs2_lock_res *lockres)
284 {
285         if (lockres->l_ops->get_osb)
286                 return lockres->l_ops->get_osb(lockres);
287
288         return (struct ocfs2_super *)lockres->l_priv;
289 }
290
291 static int ocfs2_lock_create(struct ocfs2_super *osb,
292                              struct ocfs2_lock_res *lockres,
293                              int level,
294                              int dlm_flags);
295 static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lockres,
296                                                      int wanted);
297 static void ocfs2_cluster_unlock(struct ocfs2_super *osb,
298                                  struct ocfs2_lock_res *lockres,
299                                  int level);
300 static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res *lockres);
301 static inline void ocfs2_generic_handle_convert_action(struct ocfs2_lock_res *lockres);
302 static inline void ocfs2_generic_handle_attach_action(struct ocfs2_lock_res *lockres);
303 static int ocfs2_generic_handle_bast(struct ocfs2_lock_res *lockres, int level);
304 static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
305                                         struct ocfs2_lock_res *lockres);
306 static inline void ocfs2_recover_from_dlm_error(struct ocfs2_lock_res *lockres,
307                                                 int convert);
308 #define ocfs2_log_dlm_error(_func, _stat, _lockres) do {        \
309         mlog(ML_ERROR, "Dlm error \"%s\" while calling %s on "  \
310                 "resource %s: %s\n", dlm_errname(_stat), _func, \
311                 _lockres->l_name, dlm_errmsg(_stat));           \
312 } while (0)
313 static void ocfs2_vote_on_unlock(struct ocfs2_super *osb,
314                                  struct ocfs2_lock_res *lockres);
315 static int ocfs2_meta_lock_update(struct inode *inode,
316                                   struct buffer_head **bh);
317 static void ocfs2_drop_osb_locks(struct ocfs2_super *osb);
318 static inline int ocfs2_highest_compat_lock_level(int level);
319
320 static void ocfs2_build_lock_name(enum ocfs2_lock_type type,
321                                   u64 blkno,
322                                   u32 generation,
323                                   char *name)
324 {
325         int len;
326
327         mlog_entry_void();
328
329         BUG_ON(type >= OCFS2_NUM_LOCK_TYPES);
330
331         len = snprintf(name, OCFS2_LOCK_ID_MAX_LEN, "%c%s%016llx%08x",
332                        ocfs2_lock_type_char(type), OCFS2_LOCK_ID_PAD,
333                        (long long)blkno, generation);
334
335         BUG_ON(len != (OCFS2_LOCK_ID_MAX_LEN - 1));
336
337         mlog(0, "built lock resource with name: %s\n", name);
338
339         mlog_exit_void();
340 }
341
342 static DEFINE_SPINLOCK(ocfs2_dlm_tracking_lock);
343
344 static void ocfs2_add_lockres_tracking(struct ocfs2_lock_res *res,
345                                        struct ocfs2_dlm_debug *dlm_debug)
346 {
347         mlog(0, "Add tracking for lockres %s\n", res->l_name);
348
349         spin_lock(&ocfs2_dlm_tracking_lock);
350         list_add(&res->l_debug_list, &dlm_debug->d_lockres_tracking);
351         spin_unlock(&ocfs2_dlm_tracking_lock);
352 }
353
354 static void ocfs2_remove_lockres_tracking(struct ocfs2_lock_res *res)
355 {
356         spin_lock(&ocfs2_dlm_tracking_lock);
357         if (!list_empty(&res->l_debug_list))
358                 list_del_init(&res->l_debug_list);
359         spin_unlock(&ocfs2_dlm_tracking_lock);
360 }
361
362 static void ocfs2_lock_res_init_common(struct ocfs2_super *osb,
363                                        struct ocfs2_lock_res *res,
364                                        enum ocfs2_lock_type type,
365                                        struct ocfs2_lock_res_ops *ops,
366                                        void *priv)
367 {
368         res->l_type          = type;
369         res->l_ops           = ops;
370         res->l_priv          = priv;
371
372         res->l_level         = LKM_IVMODE;
373         res->l_requested     = LKM_IVMODE;
374         res->l_blocking      = LKM_IVMODE;
375         res->l_action        = OCFS2_AST_INVALID;
376         res->l_unlock_action = OCFS2_UNLOCK_INVALID;
377
378         res->l_flags         = OCFS2_LOCK_INITIALIZED;
379
380         ocfs2_add_lockres_tracking(res, osb->osb_dlm_debug);
381 }
382
383 void ocfs2_lock_res_init_once(struct ocfs2_lock_res *res)
384 {
385         /* This also clears out the lock status block */
386         memset(res, 0, sizeof(struct ocfs2_lock_res));
387         spin_lock_init(&res->l_lock);
388         init_waitqueue_head(&res->l_event);
389         INIT_LIST_HEAD(&res->l_blocked_list);
390         INIT_LIST_HEAD(&res->l_mask_waiters);
391 }
392
393 void ocfs2_inode_lock_res_init(struct ocfs2_lock_res *res,
394                                enum ocfs2_lock_type type,
395                                unsigned int generation,
396                                struct inode *inode)
397 {
398         struct ocfs2_lock_res_ops *ops;
399
400         switch(type) {
401                 case OCFS2_LOCK_TYPE_RW:
402                         ops = &ocfs2_inode_rw_lops;
403                         break;
404                 case OCFS2_LOCK_TYPE_META:
405                         ops = &ocfs2_inode_meta_lops;
406                         break;
407                 case OCFS2_LOCK_TYPE_DATA:
408                         ops = &ocfs2_inode_data_lops;
409                         break;
410                 case OCFS2_LOCK_TYPE_OPEN:
411                         ops = &ocfs2_inode_open_lops;
412                         break;
413                 default:
414                         mlog_bug_on_msg(1, "type: %d\n", type);
415                         ops = NULL; /* thanks, gcc */
416                         break;
417         };
418
419         ocfs2_build_lock_name(type, OCFS2_I(inode)->ip_blkno,
420                               generation, res->l_name);
421         ocfs2_lock_res_init_common(OCFS2_SB(inode->i_sb), res, type, ops, inode);
422 }
423
424 static struct ocfs2_super *ocfs2_get_inode_osb(struct ocfs2_lock_res *lockres)
425 {
426         struct inode *inode = ocfs2_lock_res_inode(lockres);
427
428         return OCFS2_SB(inode->i_sb);
429 }
430
431 static __u64 ocfs2_get_dentry_lock_ino(struct ocfs2_lock_res *lockres)
432 {
433         __be64 inode_blkno_be;
434
435         memcpy(&inode_blkno_be, &lockres->l_name[OCFS2_DENTRY_LOCK_INO_START],
436                sizeof(__be64));
437
438         return be64_to_cpu(inode_blkno_be);
439 }
440
441 static struct ocfs2_super *ocfs2_get_dentry_osb(struct ocfs2_lock_res *lockres)
442 {
443         struct ocfs2_dentry_lock *dl = lockres->l_priv;
444
445         return OCFS2_SB(dl->dl_inode->i_sb);
446 }
447
448 void ocfs2_dentry_lock_res_init(struct ocfs2_dentry_lock *dl,
449                                 u64 parent, struct inode *inode)
450 {
451         int len;
452         u64 inode_blkno = OCFS2_I(inode)->ip_blkno;
453         __be64 inode_blkno_be = cpu_to_be64(inode_blkno);
454         struct ocfs2_lock_res *lockres = &dl->dl_lockres;
455
456         ocfs2_lock_res_init_once(lockres);
457
458         /*
459          * Unfortunately, the standard lock naming scheme won't work
460          * here because we have two 16 byte values to use. Instead,
461          * we'll stuff the inode number as a binary value. We still
462          * want error prints to show something without garbling the
463          * display, so drop a null byte in there before the inode
464          * number. A future version of OCFS2 will likely use all
465          * binary lock names. The stringified names have been a
466          * tremendous aid in debugging, but now that the debugfs
467          * interface exists, we can mangle things there if need be.
468          *
469          * NOTE: We also drop the standard "pad" value (the total lock
470          * name size stays the same though - the last part is all
471          * zeros due to the memset in ocfs2_lock_res_init_once()
472          */
473         len = snprintf(lockres->l_name, OCFS2_DENTRY_LOCK_INO_START,
474                        "%c%016llx",
475                        ocfs2_lock_type_char(OCFS2_LOCK_TYPE_DENTRY),
476                        (long long)parent);
477
478         BUG_ON(len != (OCFS2_DENTRY_LOCK_INO_START - 1));
479
480         memcpy(&lockres->l_name[OCFS2_DENTRY_LOCK_INO_START], &inode_blkno_be,
481                sizeof(__be64));
482
483         ocfs2_lock_res_init_common(OCFS2_SB(inode->i_sb), lockres,
484                                    OCFS2_LOCK_TYPE_DENTRY, &ocfs2_dentry_lops,
485                                    dl);
486 }
487
488 static void ocfs2_super_lock_res_init(struct ocfs2_lock_res *res,
489                                       struct ocfs2_super *osb)
490 {
491         /* Superblock lockres doesn't come from a slab so we call init
492          * once on it manually.  */
493         ocfs2_lock_res_init_once(res);
494         ocfs2_build_lock_name(OCFS2_LOCK_TYPE_SUPER, OCFS2_SUPER_BLOCK_BLKNO,
495                               0, res->l_name);
496         ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_SUPER,
497                                    &ocfs2_super_lops, osb);
498 }
499
500 static void ocfs2_rename_lock_res_init(struct ocfs2_lock_res *res,
501                                        struct ocfs2_super *osb)
502 {
503         /* Rename lockres doesn't come from a slab so we call init
504          * once on it manually.  */
505         ocfs2_lock_res_init_once(res);
506         ocfs2_build_lock_name(OCFS2_LOCK_TYPE_RENAME, 0, 0, res->l_name);
507         ocfs2_lock_res_init_common(osb, res, OCFS2_LOCK_TYPE_RENAME,
508                                    &ocfs2_rename_lops, osb);
509 }
510
511 void ocfs2_lock_res_free(struct ocfs2_lock_res *res)
512 {
513         mlog_entry_void();
514
515         if (!(res->l_flags & OCFS2_LOCK_INITIALIZED))
516                 return;
517
518         ocfs2_remove_lockres_tracking(res);
519
520         mlog_bug_on_msg(!list_empty(&res->l_blocked_list),
521                         "Lockres %s is on the blocked list\n",
522                         res->l_name);
523         mlog_bug_on_msg(!list_empty(&res->l_mask_waiters),
524                         "Lockres %s has mask waiters pending\n",
525                         res->l_name);
526         mlog_bug_on_msg(spin_is_locked(&res->l_lock),
527                         "Lockres %s is locked\n",
528                         res->l_name);
529         mlog_bug_on_msg(res->l_ro_holders,
530                         "Lockres %s has %u ro holders\n",
531                         res->l_name, res->l_ro_holders);
532         mlog_bug_on_msg(res->l_ex_holders,
533                         "Lockres %s has %u ex holders\n",
534                         res->l_name, res->l_ex_holders);
535
536         /* Need to clear out the lock status block for the dlm */
537         memset(&res->l_lksb, 0, sizeof(res->l_lksb));
538
539         res->l_flags = 0UL;
540         mlog_exit_void();
541 }
542
543 static inline void ocfs2_inc_holders(struct ocfs2_lock_res *lockres,
544                                      int level)
545 {
546         mlog_entry_void();
547
548         BUG_ON(!lockres);
549
550         switch(level) {
551         case LKM_EXMODE:
552                 lockres->l_ex_holders++;
553                 break;
554         case LKM_PRMODE:
555                 lockres->l_ro_holders++;
556                 break;
557         default:
558                 BUG();
559         }
560
561         mlog_exit_void();
562 }
563
564 static inline void ocfs2_dec_holders(struct ocfs2_lock_res *lockres,
565                                      int level)
566 {
567         mlog_entry_void();
568
569         BUG_ON(!lockres);
570
571         switch(level) {
572         case LKM_EXMODE:
573                 BUG_ON(!lockres->l_ex_holders);
574                 lockres->l_ex_holders--;
575                 break;
576         case LKM_PRMODE:
577                 BUG_ON(!lockres->l_ro_holders);
578                 lockres->l_ro_holders--;
579                 break;
580         default:
581                 BUG();
582         }
583         mlog_exit_void();
584 }
585
586 /* WARNING: This function lives in a world where the only three lock
587  * levels are EX, PR, and NL. It *will* have to be adjusted when more
588  * lock types are added. */
589 static inline int ocfs2_highest_compat_lock_level(int level)
590 {
591         int new_level = LKM_EXMODE;
592
593         if (level == LKM_EXMODE)
594                 new_level = LKM_NLMODE;
595         else if (level == LKM_PRMODE)
596                 new_level = LKM_PRMODE;
597         return new_level;
598 }
599
600 static void lockres_set_flags(struct ocfs2_lock_res *lockres,
601                               unsigned long newflags)
602 {
603         struct list_head *pos, *tmp;
604         struct ocfs2_mask_waiter *mw;
605
606         assert_spin_locked(&lockres->l_lock);
607
608         lockres->l_flags = newflags;
609
610         list_for_each_safe(pos, tmp, &lockres->l_mask_waiters) {
611                 mw = list_entry(pos, struct ocfs2_mask_waiter, mw_item);
612                 if ((lockres->l_flags & mw->mw_mask) != mw->mw_goal)
613                         continue;
614
615                 list_del_init(&mw->mw_item);
616                 mw->mw_status = 0;
617                 complete(&mw->mw_complete);
618         }
619 }
620 static void lockres_or_flags(struct ocfs2_lock_res *lockres, unsigned long or)
621 {
622         lockres_set_flags(lockres, lockres->l_flags | or);
623 }
624 static void lockres_clear_flags(struct ocfs2_lock_res *lockres,
625                                 unsigned long clear)
626 {
627         lockres_set_flags(lockres, lockres->l_flags & ~clear);
628 }
629
630 static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res *lockres)
631 {
632         mlog_entry_void();
633
634         BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
635         BUG_ON(!(lockres->l_flags & OCFS2_LOCK_ATTACHED));
636         BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BLOCKED));
637         BUG_ON(lockres->l_blocking <= LKM_NLMODE);
638
639         lockres->l_level = lockres->l_requested;
640         if (lockres->l_level <=
641             ocfs2_highest_compat_lock_level(lockres->l_blocking)) {
642                 lockres->l_blocking = LKM_NLMODE;
643                 lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED);
644         }
645         lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
646
647         mlog_exit_void();
648 }
649
650 static inline void ocfs2_generic_handle_convert_action(struct ocfs2_lock_res *lockres)
651 {
652         mlog_entry_void();
653
654         BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BUSY));
655         BUG_ON(!(lockres->l_flags & OCFS2_LOCK_ATTACHED));
656
657         /* Convert from RO to EX doesn't really need anything as our
658          * information is already up to data. Convert from NL to
659          * *anything* however should mark ourselves as needing an
660          * update */
661         if (lockres->l_level == LKM_NLMODE &&
662             lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
663                 lockres_or_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
664
665         lockres->l_level = lockres->l_requested;
666         lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
667
668         mlog_exit_void();
669 }
670
671 static inline void ocfs2_generic_handle_attach_action(struct ocfs2_lock_res *lockres)
672 {
673         mlog_entry_void();
674
675         BUG_ON((!lockres->l_flags & OCFS2_LOCK_BUSY));
676         BUG_ON(lockres->l_flags & OCFS2_LOCK_ATTACHED);
677
678         if (lockres->l_requested > LKM_NLMODE &&
679             !(lockres->l_flags & OCFS2_LOCK_LOCAL) &&
680             lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
681                 lockres_or_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
682
683         lockres->l_level = lockres->l_requested;
684         lockres_or_flags(lockres, OCFS2_LOCK_ATTACHED);
685         lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
686
687         mlog_exit_void();
688 }
689
690 static int ocfs2_generic_handle_bast(struct ocfs2_lock_res *lockres,
691                                      int level)
692 {
693         int needs_downconvert = 0;
694         mlog_entry_void();
695
696         assert_spin_locked(&lockres->l_lock);
697
698         lockres_or_flags(lockres, OCFS2_LOCK_BLOCKED);
699
700         if (level > lockres->l_blocking) {
701                 /* only schedule a downconvert if we haven't already scheduled
702                  * one that goes low enough to satisfy the level we're
703                  * blocking.  this also catches the case where we get
704                  * duplicate BASTs */
705                 if (ocfs2_highest_compat_lock_level(level) <
706                     ocfs2_highest_compat_lock_level(lockres->l_blocking))
707                         needs_downconvert = 1;
708
709                 lockres->l_blocking = level;
710         }
711
712         mlog_exit(needs_downconvert);
713         return needs_downconvert;
714 }
715
716 static void ocfs2_blocking_ast(void *opaque, int level)
717 {
718         struct ocfs2_lock_res *lockres = opaque;
719         struct ocfs2_super *osb = ocfs2_get_lockres_osb(lockres);
720         int needs_downconvert;
721         unsigned long flags;
722
723         BUG_ON(level <= LKM_NLMODE);
724
725         mlog(0, "BAST fired for lockres %s, blocking %d, level %d type %s\n",
726              lockres->l_name, level, lockres->l_level,
727              ocfs2_lock_type_string(lockres->l_type));
728
729         spin_lock_irqsave(&lockres->l_lock, flags);
730         needs_downconvert = ocfs2_generic_handle_bast(lockres, level);
731         if (needs_downconvert)
732                 ocfs2_schedule_blocked_lock(osb, lockres);
733         spin_unlock_irqrestore(&lockres->l_lock, flags);
734
735         wake_up(&lockres->l_event);
736
737         ocfs2_kick_vote_thread(osb);
738 }
739
740 static void ocfs2_locking_ast(void *opaque)
741 {
742         struct ocfs2_lock_res *lockres = opaque;
743         struct dlm_lockstatus *lksb = &lockres->l_lksb;
744         unsigned long flags;
745
746         spin_lock_irqsave(&lockres->l_lock, flags);
747
748         if (lksb->status != DLM_NORMAL) {
749                 mlog(ML_ERROR, "lockres %s: lksb status value of %u!\n",
750                      lockres->l_name, lksb->status);
751                 spin_unlock_irqrestore(&lockres->l_lock, flags);
752                 return;
753         }
754
755         switch(lockres->l_action) {
756         case OCFS2_AST_ATTACH:
757                 ocfs2_generic_handle_attach_action(lockres);
758                 lockres_clear_flags(lockres, OCFS2_LOCK_LOCAL);
759                 break;
760         case OCFS2_AST_CONVERT:
761                 ocfs2_generic_handle_convert_action(lockres);
762                 break;
763         case OCFS2_AST_DOWNCONVERT:
764                 ocfs2_generic_handle_downconvert_action(lockres);
765                 break;
766         default:
767                 mlog(ML_ERROR, "lockres %s: ast fired with invalid action: %u "
768                      "lockres flags = 0x%lx, unlock action: %u\n",
769                      lockres->l_name, lockres->l_action, lockres->l_flags,
770                      lockres->l_unlock_action);
771                 BUG();
772         }
773
774         /* set it to something invalid so if we get called again we
775          * can catch it. */
776         lockres->l_action = OCFS2_AST_INVALID;
777
778         wake_up(&lockres->l_event);
779         spin_unlock_irqrestore(&lockres->l_lock, flags);
780 }
781
782 static inline void ocfs2_recover_from_dlm_error(struct ocfs2_lock_res *lockres,
783                                                 int convert)
784 {
785         unsigned long flags;
786
787         mlog_entry_void();
788         spin_lock_irqsave(&lockres->l_lock, flags);
789         lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
790         if (convert)
791                 lockres->l_action = OCFS2_AST_INVALID;
792         else
793                 lockres->l_unlock_action = OCFS2_UNLOCK_INVALID;
794         spin_unlock_irqrestore(&lockres->l_lock, flags);
795
796         wake_up(&lockres->l_event);
797         mlog_exit_void();
798 }
799
800 /* Note: If we detect another process working on the lock (i.e.,
801  * OCFS2_LOCK_BUSY), we'll bail out returning 0. It's up to the caller
802  * to do the right thing in that case.
803  */
804 static int ocfs2_lock_create(struct ocfs2_super *osb,
805                              struct ocfs2_lock_res *lockres,
806                              int level,
807                              int dlm_flags)
808 {
809         int ret = 0;
810         enum dlm_status status = DLM_NORMAL;
811         unsigned long flags;
812
813         mlog_entry_void();
814
815         mlog(0, "lock %s, level = %d, flags = %d\n", lockres->l_name, level,
816              dlm_flags);
817
818         spin_lock_irqsave(&lockres->l_lock, flags);
819         if ((lockres->l_flags & OCFS2_LOCK_ATTACHED) ||
820             (lockres->l_flags & OCFS2_LOCK_BUSY)) {
821                 spin_unlock_irqrestore(&lockres->l_lock, flags);
822                 goto bail;
823         }
824
825         lockres->l_action = OCFS2_AST_ATTACH;
826         lockres->l_requested = level;
827         lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
828         spin_unlock_irqrestore(&lockres->l_lock, flags);
829
830         status = dlmlock(osb->dlm,
831                          level,
832                          &lockres->l_lksb,
833                          dlm_flags,
834                          lockres->l_name,
835                          OCFS2_LOCK_ID_MAX_LEN - 1,
836                          ocfs2_locking_ast,
837                          lockres,
838                          ocfs2_blocking_ast);
839         if (status != DLM_NORMAL) {
840                 ocfs2_log_dlm_error("dlmlock", status, lockres);
841                 ret = -EINVAL;
842                 ocfs2_recover_from_dlm_error(lockres, 1);
843         }
844
845         mlog(0, "lock %s, successfull return from dlmlock\n", lockres->l_name);
846
847 bail:
848         mlog_exit(ret);
849         return ret;
850 }
851
852 static inline int ocfs2_check_wait_flag(struct ocfs2_lock_res *lockres,
853                                         int flag)
854 {
855         unsigned long flags;
856         int ret;
857
858         spin_lock_irqsave(&lockres->l_lock, flags);
859         ret = lockres->l_flags & flag;
860         spin_unlock_irqrestore(&lockres->l_lock, flags);
861
862         return ret;
863 }
864
865 static inline void ocfs2_wait_on_busy_lock(struct ocfs2_lock_res *lockres)
866
867 {
868         wait_event(lockres->l_event,
869                    !ocfs2_check_wait_flag(lockres, OCFS2_LOCK_BUSY));
870 }
871
872 static inline void ocfs2_wait_on_refreshing_lock(struct ocfs2_lock_res *lockres)
873
874 {
875         wait_event(lockres->l_event,
876                    !ocfs2_check_wait_flag(lockres, OCFS2_LOCK_REFRESHING));
877 }
878
879 /* predict what lock level we'll be dropping down to on behalf
880  * of another node, and return true if the currently wanted
881  * level will be compatible with it. */
882 static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lockres,
883                                                      int wanted)
884 {
885         BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BLOCKED));
886
887         return wanted <= ocfs2_highest_compat_lock_level(lockres->l_blocking);
888 }
889
890 static void ocfs2_init_mask_waiter(struct ocfs2_mask_waiter *mw)
891 {
892         INIT_LIST_HEAD(&mw->mw_item);
893         init_completion(&mw->mw_complete);
894 }
895
896 static int ocfs2_wait_for_mask(struct ocfs2_mask_waiter *mw)
897 {
898         wait_for_completion(&mw->mw_complete);
899         /* Re-arm the completion in case we want to wait on it again */
900         INIT_COMPLETION(mw->mw_complete);
901         return mw->mw_status;
902 }
903
904 static void lockres_add_mask_waiter(struct ocfs2_lock_res *lockres,
905                                     struct ocfs2_mask_waiter *mw,
906                                     unsigned long mask,
907                                     unsigned long goal)
908 {
909         BUG_ON(!list_empty(&mw->mw_item));
910
911         assert_spin_locked(&lockres->l_lock);
912
913         list_add_tail(&mw->mw_item, &lockres->l_mask_waiters);
914         mw->mw_mask = mask;
915         mw->mw_goal = goal;
916 }
917
918 /* returns 0 if the mw that was removed was already satisfied, -EBUSY
919  * if the mask still hadn't reached its goal */
920 static int lockres_remove_mask_waiter(struct ocfs2_lock_res *lockres,
921                                       struct ocfs2_mask_waiter *mw)
922 {
923         unsigned long flags;
924         int ret = 0;
925
926         spin_lock_irqsave(&lockres->l_lock, flags);
927         if (!list_empty(&mw->mw_item)) {
928                 if ((lockres->l_flags & mw->mw_mask) != mw->mw_goal)
929                         ret = -EBUSY;
930
931                 list_del_init(&mw->mw_item);
932                 init_completion(&mw->mw_complete);
933         }
934         spin_unlock_irqrestore(&lockres->l_lock, flags);
935
936         return ret;
937
938 }
939
940 static int ocfs2_cluster_lock(struct ocfs2_super *osb,
941                               struct ocfs2_lock_res *lockres,
942                               int level,
943                               int lkm_flags,
944                               int arg_flags)
945 {
946         struct ocfs2_mask_waiter mw;
947         enum dlm_status status;
948         int wait, catch_signals = !(osb->s_mount_opt & OCFS2_MOUNT_NOINTR);
949         int ret = 0; /* gcc doesn't realize wait = 1 guarantees ret is set */
950         unsigned long flags;
951
952         mlog_entry_void();
953
954         ocfs2_init_mask_waiter(&mw);
955
956         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
957                 lkm_flags |= LKM_VALBLK;
958
959 again:
960         wait = 0;
961
962         if (catch_signals && signal_pending(current)) {
963                 ret = -ERESTARTSYS;
964                 goto out;
965         }
966
967         spin_lock_irqsave(&lockres->l_lock, flags);
968
969         mlog_bug_on_msg(lockres->l_flags & OCFS2_LOCK_FREEING,
970                         "Cluster lock called on freeing lockres %s! flags "
971                         "0x%lx\n", lockres->l_name, lockres->l_flags);
972
973         /* We only compare against the currently granted level
974          * here. If the lock is blocked waiting on a downconvert,
975          * we'll get caught below. */
976         if (lockres->l_flags & OCFS2_LOCK_BUSY &&
977             level > lockres->l_level) {
978                 /* is someone sitting in dlm_lock? If so, wait on
979                  * them. */
980                 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
981                 wait = 1;
982                 goto unlock;
983         }
984
985         if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
986                 /* lock has not been created yet. */
987                 spin_unlock_irqrestore(&lockres->l_lock, flags);
988
989                 ret = ocfs2_lock_create(osb, lockres, LKM_NLMODE, 0);
990                 if (ret < 0) {
991                         mlog_errno(ret);
992                         goto out;
993                 }
994                 goto again;
995         }
996
997         if (lockres->l_flags & OCFS2_LOCK_BLOCKED &&
998             !ocfs2_may_continue_on_blocked_lock(lockres, level)) {
999                 /* is the lock is currently blocked on behalf of
1000                  * another node */
1001                 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BLOCKED, 0);
1002                 wait = 1;
1003                 goto unlock;
1004         }
1005
1006         if (level > lockres->l_level) {
1007                 if (lockres->l_action != OCFS2_AST_INVALID)
1008                         mlog(ML_ERROR, "lockres %s has action %u pending\n",
1009                              lockres->l_name, lockres->l_action);
1010
1011                 lockres->l_action = OCFS2_AST_CONVERT;
1012                 lockres->l_requested = level;
1013                 lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
1014                 spin_unlock_irqrestore(&lockres->l_lock, flags);
1015
1016                 BUG_ON(level == LKM_IVMODE);
1017                 BUG_ON(level == LKM_NLMODE);
1018
1019                 mlog(0, "lock %s, convert from %d to level = %d\n",
1020                      lockres->l_name, lockres->l_level, level);
1021
1022                 /* call dlm_lock to upgrade lock now */
1023                 status = dlmlock(osb->dlm,
1024                                  level,
1025                                  &lockres->l_lksb,
1026                                  lkm_flags|LKM_CONVERT,
1027                                  lockres->l_name,
1028                                  OCFS2_LOCK_ID_MAX_LEN - 1,
1029                                  ocfs2_locking_ast,
1030                                  lockres,
1031                                  ocfs2_blocking_ast);
1032                 if (status != DLM_NORMAL) {
1033                         if ((lkm_flags & LKM_NOQUEUE) &&
1034                             (status == DLM_NOTQUEUED))
1035                                 ret = -EAGAIN;
1036                         else {
1037                                 ocfs2_log_dlm_error("dlmlock", status,
1038                                                     lockres);
1039                                 ret = -EINVAL;
1040                         }
1041                         ocfs2_recover_from_dlm_error(lockres, 1);
1042                         goto out;
1043                 }
1044
1045                 mlog(0, "lock %s, successfull return from dlmlock\n",
1046                      lockres->l_name);
1047
1048                 /* At this point we've gone inside the dlm and need to
1049                  * complete our work regardless. */
1050                 catch_signals = 0;
1051
1052                 /* wait for busy to clear and carry on */
1053                 goto again;
1054         }
1055
1056         /* Ok, if we get here then we're good to go. */
1057         ocfs2_inc_holders(lockres, level);
1058
1059         ret = 0;
1060 unlock:
1061         spin_unlock_irqrestore(&lockres->l_lock, flags);
1062 out:
1063         /*
1064          * This is helping work around a lock inversion between the page lock
1065          * and dlm locks.  One path holds the page lock while calling aops
1066          * which block acquiring dlm locks.  The voting thread holds dlm
1067          * locks while acquiring page locks while down converting data locks.
1068          * This block is helping an aop path notice the inversion and back
1069          * off to unlock its page lock before trying the dlm lock again.
1070          */
1071         if (wait && arg_flags & OCFS2_LOCK_NONBLOCK &&
1072             mw.mw_mask & (OCFS2_LOCK_BUSY|OCFS2_LOCK_BLOCKED)) {
1073                 wait = 0;
1074                 if (lockres_remove_mask_waiter(lockres, &mw))
1075                         ret = -EAGAIN;
1076                 else
1077                         goto again;
1078         }
1079         if (wait) {
1080                 ret = ocfs2_wait_for_mask(&mw);
1081                 if (ret == 0)
1082                         goto again;
1083                 mlog_errno(ret);
1084         }
1085
1086         mlog_exit(ret);
1087         return ret;
1088 }
1089
1090 static void ocfs2_cluster_unlock(struct ocfs2_super *osb,
1091                                  struct ocfs2_lock_res *lockres,
1092                                  int level)
1093 {
1094         unsigned long flags;
1095
1096         mlog_entry_void();
1097         spin_lock_irqsave(&lockres->l_lock, flags);
1098         ocfs2_dec_holders(lockres, level);
1099         ocfs2_vote_on_unlock(osb, lockres);
1100         spin_unlock_irqrestore(&lockres->l_lock, flags);
1101         mlog_exit_void();
1102 }
1103
1104 static int ocfs2_create_new_lock(struct ocfs2_super *osb,
1105                                  struct ocfs2_lock_res *lockres,
1106                                  int ex,
1107                                  int local)
1108 {
1109         int level =  ex ? LKM_EXMODE : LKM_PRMODE;
1110         unsigned long flags;
1111         int lkm_flags = local ? LKM_LOCAL : 0;
1112
1113         spin_lock_irqsave(&lockres->l_lock, flags);
1114         BUG_ON(lockres->l_flags & OCFS2_LOCK_ATTACHED);
1115         lockres_or_flags(lockres, OCFS2_LOCK_LOCAL);
1116         spin_unlock_irqrestore(&lockres->l_lock, flags);
1117
1118         return ocfs2_lock_create(osb, lockres, level, lkm_flags);
1119 }
1120
1121 /* Grants us an EX lock on the data and metadata resources, skipping
1122  * the normal cluster directory lookup. Use this ONLY on newly created
1123  * inodes which other nodes can't possibly see, and which haven't been
1124  * hashed in the inode hash yet. This can give us a good performance
1125  * increase as it'll skip the network broadcast normally associated
1126  * with creating a new lock resource. */
1127 int ocfs2_create_new_inode_locks(struct inode *inode)
1128 {
1129         int ret;
1130         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1131
1132         BUG_ON(!inode);
1133         BUG_ON(!ocfs2_inode_is_new(inode));
1134
1135         mlog_entry_void();
1136
1137         mlog(0, "Inode %llu\n", (unsigned long long)OCFS2_I(inode)->ip_blkno);
1138
1139         /* NOTE: That we don't increment any of the holder counts, nor
1140          * do we add anything to a journal handle. Since this is
1141          * supposed to be a new inode which the cluster doesn't know
1142          * about yet, there is no need to.  As far as the LVB handling
1143          * is concerned, this is basically like acquiring an EX lock
1144          * on a resource which has an invalid one -- we'll set it
1145          * valid when we release the EX. */
1146
1147         ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_rw_lockres, 1, 1);
1148         if (ret) {
1149                 mlog_errno(ret);
1150                 goto bail;
1151         }
1152
1153         /*
1154          * We don't want to use LKM_LOCAL on a meta data lock as they
1155          * don't use a generation in their lock names.
1156          */
1157         ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_meta_lockres, 1, 0);
1158         if (ret) {
1159                 mlog_errno(ret);
1160                 goto bail;
1161         }
1162
1163         ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_data_lockres, 1, 1);
1164         if (ret) {
1165                 mlog_errno(ret);
1166                 goto bail;
1167         }
1168
1169         ret = ocfs2_create_new_lock(osb, &OCFS2_I(inode)->ip_open_lockres, 0, 0);
1170         if (ret) {
1171                 mlog_errno(ret);
1172                 goto bail;
1173         }
1174
1175 bail:
1176         mlog_exit(ret);
1177         return ret;
1178 }
1179
1180 int ocfs2_rw_lock(struct inode *inode, int write)
1181 {
1182         int status, level;
1183         struct ocfs2_lock_res *lockres;
1184         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1185
1186         BUG_ON(!inode);
1187
1188         mlog_entry_void();
1189
1190         mlog(0, "inode %llu take %s RW lock\n",
1191              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1192              write ? "EXMODE" : "PRMODE");
1193
1194         if (ocfs2_mount_local(osb))
1195                 return 0;
1196
1197         lockres = &OCFS2_I(inode)->ip_rw_lockres;
1198
1199         level = write ? LKM_EXMODE : LKM_PRMODE;
1200
1201         status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres, level, 0,
1202                                     0);
1203         if (status < 0)
1204                 mlog_errno(status);
1205
1206         mlog_exit(status);
1207         return status;
1208 }
1209
1210 void ocfs2_rw_unlock(struct inode *inode, int write)
1211 {
1212         int level = write ? LKM_EXMODE : LKM_PRMODE;
1213         struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_rw_lockres;
1214         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1215
1216         mlog_entry_void();
1217
1218         mlog(0, "inode %llu drop %s RW lock\n",
1219              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1220              write ? "EXMODE" : "PRMODE");
1221
1222         if (!ocfs2_mount_local(osb))
1223                 ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres, level);
1224
1225         mlog_exit_void();
1226 }
1227
1228 /*
1229  * ocfs2_open_lock always get PR mode lock.
1230  */
1231 int ocfs2_open_lock(struct inode *inode)
1232 {
1233         int status = 0;
1234         struct ocfs2_lock_res *lockres;
1235         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1236
1237         BUG_ON(!inode);
1238
1239         mlog_entry_void();
1240
1241         mlog(0, "inode %llu take PRMODE open lock\n",
1242              (unsigned long long)OCFS2_I(inode)->ip_blkno);
1243
1244         if (ocfs2_mount_local(osb))
1245                 goto out;
1246
1247         lockres = &OCFS2_I(inode)->ip_open_lockres;
1248
1249         status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres,
1250                                     LKM_PRMODE, 0, 0);
1251         if (status < 0)
1252                 mlog_errno(status);
1253
1254 out:
1255         mlog_exit(status);
1256         return status;
1257 }
1258
1259 int ocfs2_try_open_lock(struct inode *inode, int write)
1260 {
1261         int status = 0, level;
1262         struct ocfs2_lock_res *lockres;
1263         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1264
1265         BUG_ON(!inode);
1266
1267         mlog_entry_void();
1268
1269         mlog(0, "inode %llu try to take %s open lock\n",
1270              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1271              write ? "EXMODE" : "PRMODE");
1272
1273         if (ocfs2_mount_local(osb))
1274                 goto out;
1275
1276         lockres = &OCFS2_I(inode)->ip_open_lockres;
1277
1278         level = write ? LKM_EXMODE : LKM_PRMODE;
1279
1280         /*
1281          * The file system may already holding a PRMODE/EXMODE open lock.
1282          * Since we pass LKM_NOQUEUE, the request won't block waiting on
1283          * other nodes and the -EAGAIN will indicate to the caller that
1284          * this inode is still in use.
1285          */
1286         status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres,
1287                                     level, LKM_NOQUEUE, 0);
1288
1289 out:
1290         mlog_exit(status);
1291         return status;
1292 }
1293
1294 /*
1295  * ocfs2_open_unlock unlock PR and EX mode open locks.
1296  */
1297 void ocfs2_open_unlock(struct inode *inode)
1298 {
1299         struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_open_lockres;
1300         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1301
1302         mlog_entry_void();
1303
1304         mlog(0, "inode %llu drop open lock\n",
1305              (unsigned long long)OCFS2_I(inode)->ip_blkno);
1306
1307         if (ocfs2_mount_local(osb))
1308                 goto out;
1309
1310         if(lockres->l_ro_holders)
1311                 ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres,
1312                                      LKM_PRMODE);
1313         if(lockres->l_ex_holders)
1314                 ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres,
1315                                      LKM_EXMODE);
1316
1317 out:
1318         mlog_exit_void();
1319 }
1320
1321 int ocfs2_data_lock_full(struct inode *inode,
1322                          int write,
1323                          int arg_flags)
1324 {
1325         int status = 0, level;
1326         struct ocfs2_lock_res *lockres;
1327         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1328
1329         BUG_ON(!inode);
1330
1331         mlog_entry_void();
1332
1333         mlog(0, "inode %llu take %s DATA lock\n",
1334              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1335              write ? "EXMODE" : "PRMODE");
1336
1337         /* We'll allow faking a readonly data lock for
1338          * rodevices. */
1339         if (ocfs2_is_hard_readonly(OCFS2_SB(inode->i_sb))) {
1340                 if (write) {
1341                         status = -EROFS;
1342                         mlog_errno(status);
1343                 }
1344                 goto out;
1345         }
1346
1347         if (ocfs2_mount_local(osb))
1348                 goto out;
1349
1350         lockres = &OCFS2_I(inode)->ip_data_lockres;
1351
1352         level = write ? LKM_EXMODE : LKM_PRMODE;
1353
1354         status = ocfs2_cluster_lock(OCFS2_SB(inode->i_sb), lockres, level,
1355                                     0, arg_flags);
1356         if (status < 0 && status != -EAGAIN)
1357                 mlog_errno(status);
1358
1359 out:
1360         mlog_exit(status);
1361         return status;
1362 }
1363
1364 /* see ocfs2_meta_lock_with_page() */
1365 int ocfs2_data_lock_with_page(struct inode *inode,
1366                               int write,
1367                               struct page *page)
1368 {
1369         int ret;
1370
1371         ret = ocfs2_data_lock_full(inode, write, OCFS2_LOCK_NONBLOCK);
1372         if (ret == -EAGAIN) {
1373                 unlock_page(page);
1374                 if (ocfs2_data_lock(inode, write) == 0)
1375                         ocfs2_data_unlock(inode, write);
1376                 ret = AOP_TRUNCATED_PAGE;
1377         }
1378
1379         return ret;
1380 }
1381
1382 static void ocfs2_vote_on_unlock(struct ocfs2_super *osb,
1383                                  struct ocfs2_lock_res *lockres)
1384 {
1385         int kick = 0;
1386
1387         mlog_entry_void();
1388
1389         /* If we know that another node is waiting on our lock, kick
1390          * the vote thread * pre-emptively when we reach a release
1391          * condition. */
1392         if (lockres->l_flags & OCFS2_LOCK_BLOCKED) {
1393                 switch(lockres->l_blocking) {
1394                 case LKM_EXMODE:
1395                         if (!lockres->l_ex_holders && !lockres->l_ro_holders)
1396                                 kick = 1;
1397                         break;
1398                 case LKM_PRMODE:
1399                         if (!lockres->l_ex_holders)
1400                                 kick = 1;
1401                         break;
1402                 default:
1403                         BUG();
1404                 }
1405         }
1406
1407         if (kick)
1408                 ocfs2_kick_vote_thread(osb);
1409
1410         mlog_exit_void();
1411 }
1412
1413 void ocfs2_data_unlock(struct inode *inode,
1414                        int write)
1415 {
1416         int level = write ? LKM_EXMODE : LKM_PRMODE;
1417         struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_data_lockres;
1418         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1419
1420         mlog_entry_void();
1421
1422         mlog(0, "inode %llu drop %s DATA lock\n",
1423              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1424              write ? "EXMODE" : "PRMODE");
1425
1426         if (!ocfs2_is_hard_readonly(OCFS2_SB(inode->i_sb)) &&
1427             !ocfs2_mount_local(osb))
1428                 ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres, level);
1429
1430         mlog_exit_void();
1431 }
1432
1433 #define OCFS2_SEC_BITS   34
1434 #define OCFS2_SEC_SHIFT  (64 - 34)
1435 #define OCFS2_NSEC_MASK  ((1ULL << OCFS2_SEC_SHIFT) - 1)
1436
1437 /* LVB only has room for 64 bits of time here so we pack it for
1438  * now. */
1439 static u64 ocfs2_pack_timespec(struct timespec *spec)
1440 {
1441         u64 res;
1442         u64 sec = spec->tv_sec;
1443         u32 nsec = spec->tv_nsec;
1444
1445         res = (sec << OCFS2_SEC_SHIFT) | (nsec & OCFS2_NSEC_MASK);
1446
1447         return res;
1448 }
1449
1450 /* Call this with the lockres locked. I am reasonably sure we don't
1451  * need ip_lock in this function as anyone who would be changing those
1452  * values is supposed to be blocked in ocfs2_meta_lock right now. */
1453 static void __ocfs2_stuff_meta_lvb(struct inode *inode)
1454 {
1455         struct ocfs2_inode_info *oi = OCFS2_I(inode);
1456         struct ocfs2_lock_res *lockres = &oi->ip_meta_lockres;
1457         struct ocfs2_meta_lvb *lvb;
1458
1459         mlog_entry_void();
1460
1461         lvb = (struct ocfs2_meta_lvb *) lockres->l_lksb.lvb;
1462
1463         /*
1464          * Invalidate the LVB of a deleted inode - this way other
1465          * nodes are forced to go to disk and discover the new inode
1466          * status.
1467          */
1468         if (oi->ip_flags & OCFS2_INODE_DELETED) {
1469                 lvb->lvb_version = 0;
1470                 goto out;
1471         }
1472
1473         lvb->lvb_version   = OCFS2_LVB_VERSION;
1474         lvb->lvb_isize     = cpu_to_be64(i_size_read(inode));
1475         lvb->lvb_iclusters = cpu_to_be32(oi->ip_clusters);
1476         lvb->lvb_iuid      = cpu_to_be32(inode->i_uid);
1477         lvb->lvb_igid      = cpu_to_be32(inode->i_gid);
1478         lvb->lvb_imode     = cpu_to_be16(inode->i_mode);
1479         lvb->lvb_inlink    = cpu_to_be16(inode->i_nlink);
1480         lvb->lvb_iatime_packed  =
1481                 cpu_to_be64(ocfs2_pack_timespec(&inode->i_atime));
1482         lvb->lvb_ictime_packed =
1483                 cpu_to_be64(ocfs2_pack_timespec(&inode->i_ctime));
1484         lvb->lvb_imtime_packed =
1485                 cpu_to_be64(ocfs2_pack_timespec(&inode->i_mtime));
1486         lvb->lvb_iattr    = cpu_to_be32(oi->ip_attr);
1487         lvb->lvb_igeneration = cpu_to_be32(inode->i_generation);
1488
1489 out:
1490         mlog_meta_lvb(0, lockres);
1491
1492         mlog_exit_void();
1493 }
1494
1495 static void ocfs2_unpack_timespec(struct timespec *spec,
1496                                   u64 packed_time)
1497 {
1498         spec->tv_sec = packed_time >> OCFS2_SEC_SHIFT;
1499         spec->tv_nsec = packed_time & OCFS2_NSEC_MASK;
1500 }
1501
1502 static void ocfs2_refresh_inode_from_lvb(struct inode *inode)
1503 {
1504         struct ocfs2_inode_info *oi = OCFS2_I(inode);
1505         struct ocfs2_lock_res *lockres = &oi->ip_meta_lockres;
1506         struct ocfs2_meta_lvb *lvb;
1507
1508         mlog_entry_void();
1509
1510         mlog_meta_lvb(0, lockres);
1511
1512         lvb = (struct ocfs2_meta_lvb *) lockres->l_lksb.lvb;
1513
1514         /* We're safe here without the lockres lock... */
1515         spin_lock(&oi->ip_lock);
1516         oi->ip_clusters = be32_to_cpu(lvb->lvb_iclusters);
1517         i_size_write(inode, be64_to_cpu(lvb->lvb_isize));
1518
1519         oi->ip_attr = be32_to_cpu(lvb->lvb_iattr);
1520         ocfs2_set_inode_flags(inode);
1521
1522         /* fast-symlinks are a special case */
1523         if (S_ISLNK(inode->i_mode) && !oi->ip_clusters)
1524                 inode->i_blocks = 0;
1525         else
1526                 inode->i_blocks = ocfs2_inode_sector_count(inode);
1527
1528         inode->i_uid     = be32_to_cpu(lvb->lvb_iuid);
1529         inode->i_gid     = be32_to_cpu(lvb->lvb_igid);
1530         inode->i_mode    = be16_to_cpu(lvb->lvb_imode);
1531         inode->i_nlink   = be16_to_cpu(lvb->lvb_inlink);
1532         ocfs2_unpack_timespec(&inode->i_atime,
1533                               be64_to_cpu(lvb->lvb_iatime_packed));
1534         ocfs2_unpack_timespec(&inode->i_mtime,
1535                               be64_to_cpu(lvb->lvb_imtime_packed));
1536         ocfs2_unpack_timespec(&inode->i_ctime,
1537                               be64_to_cpu(lvb->lvb_ictime_packed));
1538         spin_unlock(&oi->ip_lock);
1539
1540         mlog_exit_void();
1541 }
1542
1543 static inline int ocfs2_meta_lvb_is_trustable(struct inode *inode,
1544                                               struct ocfs2_lock_res *lockres)
1545 {
1546         struct ocfs2_meta_lvb *lvb = (struct ocfs2_meta_lvb *) lockres->l_lksb.lvb;
1547
1548         if (lvb->lvb_version == OCFS2_LVB_VERSION
1549             && be32_to_cpu(lvb->lvb_igeneration) == inode->i_generation)
1550                 return 1;
1551         return 0;
1552 }
1553
1554 /* Determine whether a lock resource needs to be refreshed, and
1555  * arbitrate who gets to refresh it.
1556  *
1557  *   0 means no refresh needed.
1558  *
1559  *   > 0 means you need to refresh this and you MUST call
1560  *   ocfs2_complete_lock_res_refresh afterwards. */
1561 static int ocfs2_should_refresh_lock_res(struct ocfs2_lock_res *lockres)
1562 {
1563         unsigned long flags;
1564         int status = 0;
1565
1566         mlog_entry_void();
1567
1568 refresh_check:
1569         spin_lock_irqsave(&lockres->l_lock, flags);
1570         if (!(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH)) {
1571                 spin_unlock_irqrestore(&lockres->l_lock, flags);
1572                 goto bail;
1573         }
1574
1575         if (lockres->l_flags & OCFS2_LOCK_REFRESHING) {
1576                 spin_unlock_irqrestore(&lockres->l_lock, flags);
1577
1578                 ocfs2_wait_on_refreshing_lock(lockres);
1579                 goto refresh_check;
1580         }
1581
1582         /* Ok, I'll be the one to refresh this lock. */
1583         lockres_or_flags(lockres, OCFS2_LOCK_REFRESHING);
1584         spin_unlock_irqrestore(&lockres->l_lock, flags);
1585
1586         status = 1;
1587 bail:
1588         mlog_exit(status);
1589         return status;
1590 }
1591
1592 /* If status is non zero, I'll mark it as not being in refresh
1593  * anymroe, but i won't clear the needs refresh flag. */
1594 static inline void ocfs2_complete_lock_res_refresh(struct ocfs2_lock_res *lockres,
1595                                                    int status)
1596 {
1597         unsigned long flags;
1598         mlog_entry_void();
1599
1600         spin_lock_irqsave(&lockres->l_lock, flags);
1601         lockres_clear_flags(lockres, OCFS2_LOCK_REFRESHING);
1602         if (!status)
1603                 lockres_clear_flags(lockres, OCFS2_LOCK_NEEDS_REFRESH);
1604         spin_unlock_irqrestore(&lockres->l_lock, flags);
1605
1606         wake_up(&lockres->l_event);
1607
1608         mlog_exit_void();
1609 }
1610
1611 /* may or may not return a bh if it went to disk. */
1612 static int ocfs2_meta_lock_update(struct inode *inode,
1613                                   struct buffer_head **bh)
1614 {
1615         int status = 0;
1616         struct ocfs2_inode_info *oi = OCFS2_I(inode);
1617         struct ocfs2_lock_res *lockres = &oi->ip_meta_lockres;
1618         struct ocfs2_dinode *fe;
1619         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1620
1621         mlog_entry_void();
1622
1623         if (ocfs2_mount_local(osb))
1624                 goto bail;
1625
1626         spin_lock(&oi->ip_lock);
1627         if (oi->ip_flags & OCFS2_INODE_DELETED) {
1628                 mlog(0, "Orphaned inode %llu was deleted while we "
1629                      "were waiting on a lock. ip_flags = 0x%x\n",
1630                      (unsigned long long)oi->ip_blkno, oi->ip_flags);
1631                 spin_unlock(&oi->ip_lock);
1632                 status = -ENOENT;
1633                 goto bail;
1634         }
1635         spin_unlock(&oi->ip_lock);
1636
1637         if (!ocfs2_should_refresh_lock_res(lockres))
1638                 goto bail;
1639
1640         /* This will discard any caching information we might have had
1641          * for the inode metadata. */
1642         ocfs2_metadata_cache_purge(inode);
1643
1644         ocfs2_extent_map_trunc(inode, 0);
1645
1646         if (ocfs2_meta_lvb_is_trustable(inode, lockres)) {
1647                 mlog(0, "Trusting LVB on inode %llu\n",
1648                      (unsigned long long)oi->ip_blkno);
1649                 ocfs2_refresh_inode_from_lvb(inode);
1650         } else {
1651                 /* Boo, we have to go to disk. */
1652                 /* read bh, cast, ocfs2_refresh_inode */
1653                 status = ocfs2_read_block(OCFS2_SB(inode->i_sb), oi->ip_blkno,
1654                                           bh, OCFS2_BH_CACHED, inode);
1655                 if (status < 0) {
1656                         mlog_errno(status);
1657                         goto bail_refresh;
1658                 }
1659                 fe = (struct ocfs2_dinode *) (*bh)->b_data;
1660
1661                 /* This is a good chance to make sure we're not
1662                  * locking an invalid object.
1663                  *
1664                  * We bug on a stale inode here because we checked
1665                  * above whether it was wiped from disk. The wiping
1666                  * node provides a guarantee that we receive that
1667                  * message and can mark the inode before dropping any
1668                  * locks associated with it. */
1669                 if (!OCFS2_IS_VALID_DINODE(fe)) {
1670                         OCFS2_RO_ON_INVALID_DINODE(inode->i_sb, fe);
1671                         status = -EIO;
1672                         goto bail_refresh;
1673                 }
1674                 mlog_bug_on_msg(inode->i_generation !=
1675                                 le32_to_cpu(fe->i_generation),
1676                                 "Invalid dinode %llu disk generation: %u "
1677                                 "inode->i_generation: %u\n",
1678                                 (unsigned long long)oi->ip_blkno,
1679                                 le32_to_cpu(fe->i_generation),
1680                                 inode->i_generation);
1681                 mlog_bug_on_msg(le64_to_cpu(fe->i_dtime) ||
1682                                 !(fe->i_flags & cpu_to_le32(OCFS2_VALID_FL)),
1683                                 "Stale dinode %llu dtime: %llu flags: 0x%x\n",
1684                                 (unsigned long long)oi->ip_blkno,
1685                                 (unsigned long long)le64_to_cpu(fe->i_dtime),
1686                                 le32_to_cpu(fe->i_flags));
1687
1688                 ocfs2_refresh_inode(inode, fe);
1689         }
1690
1691         status = 0;
1692 bail_refresh:
1693         ocfs2_complete_lock_res_refresh(lockres, status);
1694 bail:
1695         mlog_exit(status);
1696         return status;
1697 }
1698
1699 static int ocfs2_assign_bh(struct inode *inode,
1700                            struct buffer_head **ret_bh,
1701                            struct buffer_head *passed_bh)
1702 {
1703         int status;
1704
1705         if (passed_bh) {
1706                 /* Ok, the update went to disk for us, use the
1707                  * returned bh. */
1708                 *ret_bh = passed_bh;
1709                 get_bh(*ret_bh);
1710
1711                 return 0;
1712         }
1713
1714         status = ocfs2_read_block(OCFS2_SB(inode->i_sb),
1715                                   OCFS2_I(inode)->ip_blkno,
1716                                   ret_bh,
1717                                   OCFS2_BH_CACHED,
1718                                   inode);
1719         if (status < 0)
1720                 mlog_errno(status);
1721
1722         return status;
1723 }
1724
1725 /*
1726  * returns < 0 error if the callback will never be called, otherwise
1727  * the result of the lock will be communicated via the callback.
1728  */
1729 int ocfs2_meta_lock_full(struct inode *inode,
1730                          struct buffer_head **ret_bh,
1731                          int ex,
1732                          int arg_flags)
1733 {
1734         int status, level, dlm_flags, acquired;
1735         struct ocfs2_lock_res *lockres = NULL;
1736         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1737         struct buffer_head *local_bh = NULL;
1738
1739         BUG_ON(!inode);
1740
1741         mlog_entry_void();
1742
1743         mlog(0, "inode %llu, take %s META lock\n",
1744              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1745              ex ? "EXMODE" : "PRMODE");
1746
1747         status = 0;
1748         acquired = 0;
1749         /* We'll allow faking a readonly metadata lock for
1750          * rodevices. */
1751         if (ocfs2_is_hard_readonly(osb)) {
1752                 if (ex)
1753                         status = -EROFS;
1754                 goto bail;
1755         }
1756
1757         if (ocfs2_mount_local(osb))
1758                 goto local;
1759
1760         if (!(arg_flags & OCFS2_META_LOCK_RECOVERY))
1761                 wait_event(osb->recovery_event,
1762                            ocfs2_node_map_is_empty(osb, &osb->recovery_map));
1763
1764         lockres = &OCFS2_I(inode)->ip_meta_lockres;
1765         level = ex ? LKM_EXMODE : LKM_PRMODE;
1766         dlm_flags = 0;
1767         if (arg_flags & OCFS2_META_LOCK_NOQUEUE)
1768                 dlm_flags |= LKM_NOQUEUE;
1769
1770         status = ocfs2_cluster_lock(osb, lockres, level, dlm_flags, arg_flags);
1771         if (status < 0) {
1772                 if (status != -EAGAIN && status != -EIOCBRETRY)
1773                         mlog_errno(status);
1774                 goto bail;
1775         }
1776
1777         /* Notify the error cleanup path to drop the cluster lock. */
1778         acquired = 1;
1779
1780         /* We wait twice because a node may have died while we were in
1781          * the lower dlm layers. The second time though, we've
1782          * committed to owning this lock so we don't allow signals to
1783          * abort the operation. */
1784         if (!(arg_flags & OCFS2_META_LOCK_RECOVERY))
1785                 wait_event(osb->recovery_event,
1786                            ocfs2_node_map_is_empty(osb, &osb->recovery_map));
1787
1788 local:
1789         /*
1790          * We only see this flag if we're being called from
1791          * ocfs2_read_locked_inode(). It means we're locking an inode
1792          * which hasn't been populated yet, so clear the refresh flag
1793          * and let the caller handle it.
1794          */
1795         if (inode->i_state & I_NEW) {
1796                 status = 0;
1797                 if (lockres)
1798                         ocfs2_complete_lock_res_refresh(lockres, 0);
1799                 goto bail;
1800         }
1801
1802         /* This is fun. The caller may want a bh back, or it may
1803          * not. ocfs2_meta_lock_update definitely wants one in, but
1804          * may or may not read one, depending on what's in the
1805          * LVB. The result of all of this is that we've *only* gone to
1806          * disk if we have to, so the complexity is worthwhile. */
1807         status = ocfs2_meta_lock_update(inode, &local_bh);
1808         if (status < 0) {
1809                 if (status != -ENOENT)
1810                         mlog_errno(status);
1811                 goto bail;
1812         }
1813
1814         if (ret_bh) {
1815                 status = ocfs2_assign_bh(inode, ret_bh, local_bh);
1816                 if (status < 0) {
1817                         mlog_errno(status);
1818                         goto bail;
1819                 }
1820         }
1821
1822 bail:
1823         if (status < 0) {
1824                 if (ret_bh && (*ret_bh)) {
1825                         brelse(*ret_bh);
1826                         *ret_bh = NULL;
1827                 }
1828                 if (acquired)
1829                         ocfs2_meta_unlock(inode, ex);
1830         }
1831
1832         if (local_bh)
1833                 brelse(local_bh);
1834
1835         mlog_exit(status);
1836         return status;
1837 }
1838
1839 /*
1840  * This is working around a lock inversion between tasks acquiring DLM locks
1841  * while holding a page lock and the vote thread which blocks dlm lock acquiry
1842  * while acquiring page locks.
1843  *
1844  * ** These _with_page variantes are only intended to be called from aop
1845  * methods that hold page locks and return a very specific *positive* error
1846  * code that aop methods pass up to the VFS -- test for errors with != 0. **
1847  *
1848  * The DLM is called such that it returns -EAGAIN if it would have blocked
1849  * waiting for the vote thread.  In that case we unlock our page so the vote
1850  * thread can make progress.  Once we've done this we have to return
1851  * AOP_TRUNCATED_PAGE so the aop method that called us can bubble that back up
1852  * into the VFS who will then immediately retry the aop call.
1853  *
1854  * We do a blocking lock and immediate unlock before returning, though, so that
1855  * the lock has a great chance of being cached on this node by the time the VFS
1856  * calls back to retry the aop.    This has a potential to livelock as nodes
1857  * ping locks back and forth, but that's a risk we're willing to take to avoid
1858  * the lock inversion simply.
1859  */
1860 int ocfs2_meta_lock_with_page(struct inode *inode,
1861                               struct buffer_head **ret_bh,
1862                               int ex,
1863                               struct page *page)
1864 {
1865         int ret;
1866
1867         ret = ocfs2_meta_lock_full(inode, ret_bh, ex, OCFS2_LOCK_NONBLOCK);
1868         if (ret == -EAGAIN) {
1869                 unlock_page(page);
1870                 if (ocfs2_meta_lock(inode, ret_bh, ex) == 0)
1871                         ocfs2_meta_unlock(inode, ex);
1872                 ret = AOP_TRUNCATED_PAGE;
1873         }
1874
1875         return ret;
1876 }
1877
1878 int ocfs2_meta_lock_atime(struct inode *inode,
1879                           struct vfsmount *vfsmnt,
1880                           int *level)
1881 {
1882         int ret;
1883
1884         mlog_entry_void();
1885         ret = ocfs2_meta_lock(inode, NULL, 0);
1886         if (ret < 0) {
1887                 mlog_errno(ret);
1888                 return ret;
1889         }
1890
1891         /*
1892          * If we should update atime, we will get EX lock,
1893          * otherwise we just get PR lock.
1894          */
1895         if (ocfs2_should_update_atime(inode, vfsmnt)) {
1896                 struct buffer_head *bh = NULL;
1897
1898                 ocfs2_meta_unlock(inode, 0);
1899                 ret = ocfs2_meta_lock(inode, &bh, 1);
1900                 if (ret < 0) {
1901                         mlog_errno(ret);
1902                         return ret;
1903                 }
1904                 *level = 1;
1905                 if (ocfs2_should_update_atime(inode, vfsmnt))
1906                         ocfs2_update_inode_atime(inode, bh);
1907                 if (bh)
1908                         brelse(bh);
1909         } else
1910                 *level = 0;
1911
1912         mlog_exit(ret);
1913         return ret;
1914 }
1915
1916 void ocfs2_meta_unlock(struct inode *inode,
1917                        int ex)
1918 {
1919         int level = ex ? LKM_EXMODE : LKM_PRMODE;
1920         struct ocfs2_lock_res *lockres = &OCFS2_I(inode)->ip_meta_lockres;
1921         struct ocfs2_super *osb = OCFS2_SB(inode->i_sb);
1922
1923         mlog_entry_void();
1924
1925         mlog(0, "inode %llu drop %s META lock\n",
1926              (unsigned long long)OCFS2_I(inode)->ip_blkno,
1927              ex ? "EXMODE" : "PRMODE");
1928
1929         if (!ocfs2_is_hard_readonly(OCFS2_SB(inode->i_sb)) &&
1930             !ocfs2_mount_local(osb))
1931                 ocfs2_cluster_unlock(OCFS2_SB(inode->i_sb), lockres, level);
1932
1933         mlog_exit_void();
1934 }
1935
1936 int ocfs2_super_lock(struct ocfs2_super *osb,
1937                      int ex)
1938 {
1939         int status = 0;
1940         int level = ex ? LKM_EXMODE : LKM_PRMODE;
1941         struct ocfs2_lock_res *lockres = &osb->osb_super_lockres;
1942         struct buffer_head *bh;
1943         struct ocfs2_slot_info *si = osb->slot_info;
1944
1945         mlog_entry_void();
1946
1947         if (ocfs2_is_hard_readonly(osb))
1948                 return -EROFS;
1949
1950         if (ocfs2_mount_local(osb))
1951                 goto bail;
1952
1953         status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
1954         if (status < 0) {
1955                 mlog_errno(status);
1956                 goto bail;
1957         }
1958
1959         /* The super block lock path is really in the best position to
1960          * know when resources covered by the lock need to be
1961          * refreshed, so we do it here. Of course, making sense of
1962          * everything is up to the caller :) */
1963         status = ocfs2_should_refresh_lock_res(lockres);
1964         if (status < 0) {
1965                 mlog_errno(status);
1966                 goto bail;
1967         }
1968         if (status) {
1969                 bh = si->si_bh;
1970                 status = ocfs2_read_block(osb, bh->b_blocknr, &bh, 0,
1971                                           si->si_inode);
1972                 if (status == 0)
1973                         ocfs2_update_slot_info(si);
1974
1975                 ocfs2_complete_lock_res_refresh(lockres, status);
1976
1977                 if (status < 0)
1978                         mlog_errno(status);
1979         }
1980 bail:
1981         mlog_exit(status);
1982         return status;
1983 }
1984
1985 void ocfs2_super_unlock(struct ocfs2_super *osb,
1986                         int ex)
1987 {
1988         int level = ex ? LKM_EXMODE : LKM_PRMODE;
1989         struct ocfs2_lock_res *lockres = &osb->osb_super_lockres;
1990
1991         if (!ocfs2_mount_local(osb))
1992                 ocfs2_cluster_unlock(osb, lockres, level);
1993 }
1994
1995 int ocfs2_rename_lock(struct ocfs2_super *osb)
1996 {
1997         int status;
1998         struct ocfs2_lock_res *lockres = &osb->osb_rename_lockres;
1999
2000         if (ocfs2_is_hard_readonly(osb))
2001                 return -EROFS;
2002
2003         if (ocfs2_mount_local(osb))
2004                 return 0;
2005
2006         status = ocfs2_cluster_lock(osb, lockres, LKM_EXMODE, 0, 0);
2007         if (status < 0)
2008                 mlog_errno(status);
2009
2010         return status;
2011 }
2012
2013 void ocfs2_rename_unlock(struct ocfs2_super *osb)
2014 {
2015         struct ocfs2_lock_res *lockres = &osb->osb_rename_lockres;
2016
2017         if (!ocfs2_mount_local(osb))
2018                 ocfs2_cluster_unlock(osb, lockres, LKM_EXMODE);
2019 }
2020
2021 int ocfs2_dentry_lock(struct dentry *dentry, int ex)
2022 {
2023         int ret;
2024         int level = ex ? LKM_EXMODE : LKM_PRMODE;
2025         struct ocfs2_dentry_lock *dl = dentry->d_fsdata;
2026         struct ocfs2_super *osb = OCFS2_SB(dentry->d_sb);
2027
2028         BUG_ON(!dl);
2029
2030         if (ocfs2_is_hard_readonly(osb))
2031                 return -EROFS;
2032
2033         if (ocfs2_mount_local(osb))
2034                 return 0;
2035
2036         ret = ocfs2_cluster_lock(osb, &dl->dl_lockres, level, 0, 0);
2037         if (ret < 0)
2038                 mlog_errno(ret);
2039
2040         return ret;
2041 }
2042
2043 void ocfs2_dentry_unlock(struct dentry *dentry, int ex)
2044 {
2045         int level = ex ? LKM_EXMODE : LKM_PRMODE;
2046         struct ocfs2_dentry_lock *dl = dentry->d_fsdata;
2047         struct ocfs2_super *osb = OCFS2_SB(dentry->d_sb);
2048
2049         if (!ocfs2_mount_local(osb))
2050                 ocfs2_cluster_unlock(osb, &dl->dl_lockres, level);
2051 }
2052
2053 /* Reference counting of the dlm debug structure. We want this because
2054  * open references on the debug inodes can live on after a mount, so
2055  * we can't rely on the ocfs2_super to always exist. */
2056 static void ocfs2_dlm_debug_free(struct kref *kref)
2057 {
2058         struct ocfs2_dlm_debug *dlm_debug;
2059
2060         dlm_debug = container_of(kref, struct ocfs2_dlm_debug, d_refcnt);
2061
2062         kfree(dlm_debug);
2063 }
2064
2065 void ocfs2_put_dlm_debug(struct ocfs2_dlm_debug *dlm_debug)
2066 {
2067         if (dlm_debug)
2068                 kref_put(&dlm_debug->d_refcnt, ocfs2_dlm_debug_free);
2069 }
2070
2071 static void ocfs2_get_dlm_debug(struct ocfs2_dlm_debug *debug)
2072 {
2073         kref_get(&debug->d_refcnt);
2074 }
2075
2076 struct ocfs2_dlm_debug *ocfs2_new_dlm_debug(void)
2077 {
2078         struct ocfs2_dlm_debug *dlm_debug;
2079
2080         dlm_debug = kmalloc(sizeof(struct ocfs2_dlm_debug), GFP_KERNEL);
2081         if (!dlm_debug) {
2082                 mlog_errno(-ENOMEM);
2083                 goto out;
2084         }
2085
2086         kref_init(&dlm_debug->d_refcnt);
2087         INIT_LIST_HEAD(&dlm_debug->d_lockres_tracking);
2088         dlm_debug->d_locking_state = NULL;
2089 out:
2090         return dlm_debug;
2091 }
2092
2093 /* Access to this is arbitrated for us via seq_file->sem. */
2094 struct ocfs2_dlm_seq_priv {
2095         struct ocfs2_dlm_debug *p_dlm_debug;
2096         struct ocfs2_lock_res p_iter_res;
2097         struct ocfs2_lock_res p_tmp_res;
2098 };
2099
2100 static struct ocfs2_lock_res *ocfs2_dlm_next_res(struct ocfs2_lock_res *start,
2101                                                  struct ocfs2_dlm_seq_priv *priv)
2102 {
2103         struct ocfs2_lock_res *iter, *ret = NULL;
2104         struct ocfs2_dlm_debug *dlm_debug = priv->p_dlm_debug;
2105
2106         assert_spin_locked(&ocfs2_dlm_tracking_lock);
2107
2108         list_for_each_entry(iter, &start->l_debug_list, l_debug_list) {
2109                 /* discover the head of the list */
2110                 if (&iter->l_debug_list == &dlm_debug->d_lockres_tracking) {
2111                         mlog(0, "End of list found, %p\n", ret);
2112                         break;
2113                 }
2114
2115                 /* We track our "dummy" iteration lockres' by a NULL
2116                  * l_ops field. */
2117                 if (iter->l_ops != NULL) {
2118                         ret = iter;
2119                         break;
2120                 }
2121         }
2122
2123         return ret;
2124 }
2125
2126 static void *ocfs2_dlm_seq_start(struct seq_file *m, loff_t *pos)
2127 {
2128         struct ocfs2_dlm_seq_priv *priv = m->private;
2129         struct ocfs2_lock_res *iter;
2130
2131         spin_lock(&ocfs2_dlm_tracking_lock);
2132         iter = ocfs2_dlm_next_res(&priv->p_iter_res, priv);
2133         if (iter) {
2134                 /* Since lockres' have the lifetime of their container
2135                  * (which can be inodes, ocfs2_supers, etc) we want to
2136                  * copy this out to a temporary lockres while still
2137                  * under the spinlock. Obviously after this we can't
2138                  * trust any pointers on the copy returned, but that's
2139                  * ok as the information we want isn't typically held
2140                  * in them. */
2141                 priv->p_tmp_res = *iter;
2142                 iter = &priv->p_tmp_res;
2143         }
2144         spin_unlock(&ocfs2_dlm_tracking_lock);
2145
2146         return iter;
2147 }
2148
2149 static void ocfs2_dlm_seq_stop(struct seq_file *m, void *v)
2150 {
2151 }
2152
2153 static void *ocfs2_dlm_seq_next(struct seq_file *m, void *v, loff_t *pos)
2154 {
2155         struct ocfs2_dlm_seq_priv *priv = m->private;
2156         struct ocfs2_lock_res *iter = v;
2157         struct ocfs2_lock_res *dummy = &priv->p_iter_res;
2158
2159         spin_lock(&ocfs2_dlm_tracking_lock);
2160         iter = ocfs2_dlm_next_res(iter, priv);
2161         list_del_init(&dummy->l_debug_list);
2162         if (iter) {
2163                 list_add(&dummy->l_debug_list, &iter->l_debug_list);
2164                 priv->p_tmp_res = *iter;
2165                 iter = &priv->p_tmp_res;
2166         }
2167         spin_unlock(&ocfs2_dlm_tracking_lock);
2168
2169         return iter;
2170 }
2171
2172 /* So that debugfs.ocfs2 can determine which format is being used */
2173 #define OCFS2_DLM_DEBUG_STR_VERSION 1
2174 static int ocfs2_dlm_seq_show(struct seq_file *m, void *v)
2175 {
2176         int i;
2177         char *lvb;
2178         struct ocfs2_lock_res *lockres = v;
2179
2180         if (!lockres)
2181                 return -EINVAL;
2182
2183         seq_printf(m, "0x%x\t", OCFS2_DLM_DEBUG_STR_VERSION);
2184
2185         if (lockres->l_type == OCFS2_LOCK_TYPE_DENTRY)
2186                 seq_printf(m, "%.*s%08x\t", OCFS2_DENTRY_LOCK_INO_START - 1,
2187                            lockres->l_name,
2188                            (unsigned int)ocfs2_get_dentry_lock_ino(lockres));
2189         else
2190                 seq_printf(m, "%.*s\t", OCFS2_LOCK_ID_MAX_LEN, lockres->l_name);
2191
2192         seq_printf(m, "%d\t"
2193                    "0x%lx\t"
2194                    "0x%x\t"
2195                    "0x%x\t"
2196                    "%u\t"
2197                    "%u\t"
2198                    "%d\t"
2199                    "%d\t",
2200                    lockres->l_level,
2201                    lockres->l_flags,
2202                    lockres->l_action,
2203                    lockres->l_unlock_action,
2204                    lockres->l_ro_holders,
2205                    lockres->l_ex_holders,
2206                    lockres->l_requested,
2207                    lockres->l_blocking);
2208
2209         /* Dump the raw LVB */
2210         lvb = lockres->l_lksb.lvb;
2211         for(i = 0; i < DLM_LVB_LEN; i++)
2212                 seq_printf(m, "0x%x\t", lvb[i]);
2213
2214         /* End the line */
2215         seq_printf(m, "\n");
2216         return 0;
2217 }
2218
2219 static struct seq_operations ocfs2_dlm_seq_ops = {
2220         .start =        ocfs2_dlm_seq_start,
2221         .stop =         ocfs2_dlm_seq_stop,
2222         .next =         ocfs2_dlm_seq_next,
2223         .show =         ocfs2_dlm_seq_show,
2224 };
2225
2226 static int ocfs2_dlm_debug_release(struct inode *inode, struct file *file)
2227 {
2228         struct seq_file *seq = (struct seq_file *) file->private_data;
2229         struct ocfs2_dlm_seq_priv *priv = seq->private;
2230         struct ocfs2_lock_res *res = &priv->p_iter_res;
2231
2232         ocfs2_remove_lockres_tracking(res);
2233         ocfs2_put_dlm_debug(priv->p_dlm_debug);
2234         return seq_release_private(inode, file);
2235 }
2236
2237 static int ocfs2_dlm_debug_open(struct inode *inode, struct file *file)
2238 {
2239         int ret;
2240         struct ocfs2_dlm_seq_priv *priv;
2241         struct seq_file *seq;
2242         struct ocfs2_super *osb;
2243
2244         priv = kzalloc(sizeof(struct ocfs2_dlm_seq_priv), GFP_KERNEL);
2245         if (!priv) {
2246                 ret = -ENOMEM;
2247                 mlog_errno(ret);
2248                 goto out;
2249         }
2250         osb = inode->i_private;
2251         ocfs2_get_dlm_debug(osb->osb_dlm_debug);
2252         priv->p_dlm_debug = osb->osb_dlm_debug;
2253         INIT_LIST_HEAD(&priv->p_iter_res.l_debug_list);
2254
2255         ret = seq_open(file, &ocfs2_dlm_seq_ops);
2256         if (ret) {
2257                 kfree(priv);
2258                 mlog_errno(ret);
2259                 goto out;
2260         }
2261
2262         seq = (struct seq_file *) file->private_data;
2263         seq->private = priv;
2264
2265         ocfs2_add_lockres_tracking(&priv->p_iter_res,
2266                                    priv->p_dlm_debug);
2267
2268 out:
2269         return ret;
2270 }
2271
2272 static const struct file_operations ocfs2_dlm_debug_fops = {
2273         .open =         ocfs2_dlm_debug_open,
2274         .release =      ocfs2_dlm_debug_release,
2275         .read =         seq_read,
2276         .llseek =       seq_lseek,
2277 };
2278
2279 static int ocfs2_dlm_init_debug(struct ocfs2_super *osb)
2280 {
2281         int ret = 0;
2282         struct ocfs2_dlm_debug *dlm_debug = osb->osb_dlm_debug;
2283
2284         dlm_debug->d_locking_state = debugfs_create_file("locking_state",
2285                                                          S_IFREG|S_IRUSR,
2286                                                          osb->osb_debug_root,
2287                                                          osb,
2288                                                          &ocfs2_dlm_debug_fops);
2289         if (!dlm_debug->d_locking_state) {
2290                 ret = -EINVAL;
2291                 mlog(ML_ERROR,
2292                      "Unable to create locking state debugfs file.\n");
2293                 goto out;
2294         }
2295
2296         ocfs2_get_dlm_debug(dlm_debug);
2297 out:
2298         return ret;
2299 }
2300
2301 static void ocfs2_dlm_shutdown_debug(struct ocfs2_super *osb)
2302 {
2303         struct ocfs2_dlm_debug *dlm_debug = osb->osb_dlm_debug;
2304
2305         if (dlm_debug) {
2306                 debugfs_remove(dlm_debug->d_locking_state);
2307                 ocfs2_put_dlm_debug(dlm_debug);
2308         }
2309 }
2310
2311 int ocfs2_dlm_init(struct ocfs2_super *osb)
2312 {
2313         int status = 0;
2314         u32 dlm_key;
2315         struct dlm_ctxt *dlm = NULL;
2316
2317         mlog_entry_void();
2318
2319         if (ocfs2_mount_local(osb))
2320                 goto local;
2321
2322         status = ocfs2_dlm_init_debug(osb);
2323         if (status < 0) {
2324                 mlog_errno(status);
2325                 goto bail;
2326         }
2327
2328         /* launch vote thread */
2329         osb->vote_task = kthread_run(ocfs2_vote_thread, osb, "ocfs2vote");
2330         if (IS_ERR(osb->vote_task)) {
2331                 status = PTR_ERR(osb->vote_task);
2332                 osb->vote_task = NULL;
2333                 mlog_errno(status);
2334                 goto bail;
2335         }
2336
2337         /* used by the dlm code to make message headers unique, each
2338          * node in this domain must agree on this. */
2339         dlm_key = crc32_le(0, osb->uuid_str, strlen(osb->uuid_str));
2340
2341         /* for now, uuid == domain */
2342         dlm = dlm_register_domain(osb->uuid_str, dlm_key);
2343         if (IS_ERR(dlm)) {
2344                 status = PTR_ERR(dlm);
2345                 mlog_errno(status);
2346                 goto bail;
2347         }
2348
2349         dlm_register_eviction_cb(dlm, &osb->osb_eviction_cb);
2350
2351 local:
2352         ocfs2_super_lock_res_init(&osb->osb_super_lockres, osb);
2353         ocfs2_rename_lock_res_init(&osb->osb_rename_lockres, osb);
2354
2355         osb->dlm = dlm;
2356
2357         status = 0;
2358 bail:
2359         if (status < 0) {
2360                 ocfs2_dlm_shutdown_debug(osb);
2361                 if (osb->vote_task)
2362                         kthread_stop(osb->vote_task);
2363         }
2364
2365         mlog_exit(status);
2366         return status;
2367 }
2368
2369 void ocfs2_dlm_shutdown(struct ocfs2_super *osb)
2370 {
2371         mlog_entry_void();
2372
2373         dlm_unregister_eviction_cb(&osb->osb_eviction_cb);
2374
2375         ocfs2_drop_osb_locks(osb);
2376
2377         if (osb->vote_task) {
2378                 kthread_stop(osb->vote_task);
2379                 osb->vote_task = NULL;
2380         }
2381
2382         ocfs2_lock_res_free(&osb->osb_super_lockres);
2383         ocfs2_lock_res_free(&osb->osb_rename_lockres);
2384
2385         dlm_unregister_domain(osb->dlm);
2386         osb->dlm = NULL;
2387
2388         ocfs2_dlm_shutdown_debug(osb);
2389
2390         mlog_exit_void();
2391 }
2392
2393 static void ocfs2_unlock_ast(void *opaque, enum dlm_status status)
2394 {
2395         struct ocfs2_lock_res *lockres = opaque;
2396         unsigned long flags;
2397
2398         mlog_entry_void();
2399
2400         mlog(0, "UNLOCK AST called on lock %s, action = %d\n", lockres->l_name,
2401              lockres->l_unlock_action);
2402
2403         spin_lock_irqsave(&lockres->l_lock, flags);
2404         /* We tried to cancel a convert request, but it was already
2405          * granted. All we want to do here is clear our unlock
2406          * state. The wake_up call done at the bottom is redundant
2407          * (ocfs2_prepare_cancel_convert doesn't sleep on this) but doesn't
2408          * hurt anything anyway */
2409         if (status == DLM_CANCELGRANT &&
2410             lockres->l_unlock_action == OCFS2_UNLOCK_CANCEL_CONVERT) {
2411                 mlog(0, "Got cancelgrant for %s\n", lockres->l_name);
2412
2413                 /* We don't clear the busy flag in this case as it
2414                  * should have been cleared by the ast which the dlm
2415                  * has called. */
2416                 goto complete_unlock;
2417         }
2418
2419         if (status != DLM_NORMAL) {
2420                 mlog(ML_ERROR, "Dlm passes status %d for lock %s, "
2421                      "unlock_action %d\n", status, lockres->l_name,
2422                      lockres->l_unlock_action);
2423                 spin_unlock_irqrestore(&lockres->l_lock, flags);
2424                 return;
2425         }
2426
2427         switch(lockres->l_unlock_action) {
2428         case OCFS2_UNLOCK_CANCEL_CONVERT:
2429                 mlog(0, "Cancel convert success for %s\n", lockres->l_name);
2430                 lockres->l_action = OCFS2_AST_INVALID;
2431                 break;
2432         case OCFS2_UNLOCK_DROP_LOCK:
2433                 lockres->l_level = LKM_IVMODE;
2434                 break;
2435         default:
2436                 BUG();
2437         }
2438
2439         lockres_clear_flags(lockres, OCFS2_LOCK_BUSY);
2440 complete_unlock:
2441         lockres->l_unlock_action = OCFS2_UNLOCK_INVALID;
2442         spin_unlock_irqrestore(&lockres->l_lock, flags);
2443
2444         wake_up(&lockres->l_event);
2445
2446         mlog_exit_void();
2447 }
2448
2449 static int ocfs2_drop_lock(struct ocfs2_super *osb,
2450                            struct ocfs2_lock_res *lockres)
2451 {
2452         enum dlm_status status;
2453         unsigned long flags;
2454         int lkm_flags = 0;
2455
2456         /* We didn't get anywhere near actually using this lockres. */
2457         if (!(lockres->l_flags & OCFS2_LOCK_INITIALIZED))
2458                 goto out;
2459
2460         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
2461                 lkm_flags |= LKM_VALBLK;
2462
2463         spin_lock_irqsave(&lockres->l_lock, flags);
2464
2465         mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_FREEING),
2466                         "lockres %s, flags 0x%lx\n",
2467                         lockres->l_name, lockres->l_flags);
2468
2469         while (lockres->l_flags & OCFS2_LOCK_BUSY) {
2470                 mlog(0, "waiting on busy lock \"%s\": flags = %lx, action = "
2471                      "%u, unlock_action = %u\n",
2472                      lockres->l_name, lockres->l_flags, lockres->l_action,
2473                      lockres->l_unlock_action);
2474
2475                 spin_unlock_irqrestore(&lockres->l_lock, flags);
2476
2477                 /* XXX: Today we just wait on any busy
2478                  * locks... Perhaps we need to cancel converts in the
2479                  * future? */
2480                 ocfs2_wait_on_busy_lock(lockres);
2481
2482                 spin_lock_irqsave(&lockres->l_lock, flags);
2483         }
2484
2485         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
2486                 if (lockres->l_flags & OCFS2_LOCK_ATTACHED &&
2487                     lockres->l_level == LKM_EXMODE &&
2488                     !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
2489                         lockres->l_ops->set_lvb(lockres);
2490         }
2491
2492         if (lockres->l_flags & OCFS2_LOCK_BUSY)
2493                 mlog(ML_ERROR, "destroying busy lock: \"%s\"\n",
2494                      lockres->l_name);
2495         if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
2496                 mlog(0, "destroying blocked lock: \"%s\"\n", lockres->l_name);
2497
2498         if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
2499                 spin_unlock_irqrestore(&lockres->l_lock, flags);
2500                 goto out;
2501         }
2502
2503         lockres_clear_flags(lockres, OCFS2_LOCK_ATTACHED);
2504
2505         /* make sure we never get here while waiting for an ast to
2506          * fire. */
2507         BUG_ON(lockres->l_action != OCFS2_AST_INVALID);
2508
2509         /* is this necessary? */
2510         lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
2511         lockres->l_unlock_action = OCFS2_UNLOCK_DROP_LOCK;
2512         spin_unlock_irqrestore(&lockres->l_lock, flags);
2513
2514         mlog(0, "lock %s\n", lockres->l_name);
2515
2516         status = dlmunlock(osb->dlm, &lockres->l_lksb, lkm_flags,
2517                            ocfs2_unlock_ast, lockres);
2518         if (status != DLM_NORMAL) {
2519                 ocfs2_log_dlm_error("dlmunlock", status, lockres);
2520                 mlog(ML_ERROR, "lockres flags: %lu\n", lockres->l_flags);
2521                 dlm_print_one_lock(lockres->l_lksb.lockid);
2522                 BUG();
2523         }
2524         mlog(0, "lock %s, successfull return from dlmunlock\n",
2525              lockres->l_name);
2526
2527         ocfs2_wait_on_busy_lock(lockres);
2528 out:
2529         mlog_exit(0);
2530         return 0;
2531 }
2532
2533 /* Mark the lockres as being dropped. It will no longer be
2534  * queued if blocking, but we still may have to wait on it
2535  * being dequeued from the vote thread before we can consider
2536  * it safe to drop. 
2537  *
2538  * You can *not* attempt to call cluster_lock on this lockres anymore. */
2539 void ocfs2_mark_lockres_freeing(struct ocfs2_lock_res *lockres)
2540 {
2541         int status;
2542         struct ocfs2_mask_waiter mw;
2543         unsigned long flags;
2544
2545         ocfs2_init_mask_waiter(&mw);
2546
2547         spin_lock_irqsave(&lockres->l_lock, flags);
2548         lockres->l_flags |= OCFS2_LOCK_FREEING;
2549         while (lockres->l_flags & OCFS2_LOCK_QUEUED) {
2550                 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_QUEUED, 0);
2551                 spin_unlock_irqrestore(&lockres->l_lock, flags);
2552
2553                 mlog(0, "Waiting on lockres %s\n", lockres->l_name);
2554
2555                 status = ocfs2_wait_for_mask(&mw);
2556                 if (status)
2557                         mlog_errno(status);
2558
2559                 spin_lock_irqsave(&lockres->l_lock, flags);
2560         }
2561         spin_unlock_irqrestore(&lockres->l_lock, flags);
2562 }
2563
2564 void ocfs2_simple_drop_lockres(struct ocfs2_super *osb,
2565                                struct ocfs2_lock_res *lockres)
2566 {
2567         int ret;
2568
2569         ocfs2_mark_lockres_freeing(lockres);
2570         ret = ocfs2_drop_lock(osb, lockres);
2571         if (ret)
2572                 mlog_errno(ret);
2573 }
2574
2575 static void ocfs2_drop_osb_locks(struct ocfs2_super *osb)
2576 {
2577         ocfs2_simple_drop_lockres(osb, &osb->osb_super_lockres);
2578         ocfs2_simple_drop_lockres(osb, &osb->osb_rename_lockres);
2579 }
2580
2581 int ocfs2_drop_inode_locks(struct inode *inode)
2582 {
2583         int status, err;
2584
2585         mlog_entry_void();
2586
2587         /* No need to call ocfs2_mark_lockres_freeing here -
2588          * ocfs2_clear_inode has done it for us. */
2589
2590         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
2591                               &OCFS2_I(inode)->ip_open_lockres);
2592         if (err < 0)
2593                 mlog_errno(err);
2594
2595         status = err;
2596
2597         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
2598                               &OCFS2_I(inode)->ip_data_lockres);
2599         if (err < 0)
2600                 mlog_errno(err);
2601         if (err < 0 && !status)
2602                 status = err;
2603
2604         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
2605                               &OCFS2_I(inode)->ip_meta_lockres);
2606         if (err < 0)
2607                 mlog_errno(err);
2608         if (err < 0 && !status)
2609                 status = err;
2610
2611         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
2612                               &OCFS2_I(inode)->ip_rw_lockres);
2613         if (err < 0)
2614                 mlog_errno(err);
2615         if (err < 0 && !status)
2616                 status = err;
2617
2618         mlog_exit(status);
2619         return status;
2620 }
2621
2622 static void ocfs2_prepare_downconvert(struct ocfs2_lock_res *lockres,
2623                                       int new_level)
2624 {
2625         assert_spin_locked(&lockres->l_lock);
2626
2627         BUG_ON(lockres->l_blocking <= LKM_NLMODE);
2628
2629         if (lockres->l_level <= new_level) {
2630                 mlog(ML_ERROR, "lockres->l_level (%u) <= new_level (%u)\n",
2631                      lockres->l_level, new_level);
2632                 BUG();
2633         }
2634
2635         mlog(0, "lock %s, new_level = %d, l_blocking = %d\n",
2636              lockres->l_name, new_level, lockres->l_blocking);
2637
2638         lockres->l_action = OCFS2_AST_DOWNCONVERT;
2639         lockres->l_requested = new_level;
2640         lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
2641 }
2642
2643 static int ocfs2_downconvert_lock(struct ocfs2_super *osb,
2644                                   struct ocfs2_lock_res *lockres,
2645                                   int new_level,
2646                                   int lvb)
2647 {
2648         int ret, dlm_flags = LKM_CONVERT;
2649         enum dlm_status status;
2650
2651         mlog_entry_void();
2652
2653         if (lvb)
2654                 dlm_flags |= LKM_VALBLK;
2655
2656         status = dlmlock(osb->dlm,
2657                          new_level,
2658                          &lockres->l_lksb,
2659                          dlm_flags,
2660                          lockres->l_name,
2661                          OCFS2_LOCK_ID_MAX_LEN - 1,
2662                          ocfs2_locking_ast,
2663                          lockres,
2664                          ocfs2_blocking_ast);
2665         if (status != DLM_NORMAL) {
2666                 ocfs2_log_dlm_error("dlmlock", status, lockres);
2667                 ret = -EINVAL;
2668                 ocfs2_recover_from_dlm_error(lockres, 1);
2669                 goto bail;
2670         }
2671
2672         ret = 0;
2673 bail:
2674         mlog_exit(ret);
2675         return ret;
2676 }
2677
2678 /* returns 1 when the caller should unlock and call dlmunlock */
2679 static int ocfs2_prepare_cancel_convert(struct ocfs2_super *osb,
2680                                         struct ocfs2_lock_res *lockres)
2681 {
2682         assert_spin_locked(&lockres->l_lock);
2683
2684         mlog_entry_void();
2685         mlog(0, "lock %s\n", lockres->l_name);
2686
2687         if (lockres->l_unlock_action == OCFS2_UNLOCK_CANCEL_CONVERT) {
2688                 /* If we're already trying to cancel a lock conversion
2689                  * then just drop the spinlock and allow the caller to
2690                  * requeue this lock. */
2691
2692                 mlog(0, "Lockres %s, skip convert\n", lockres->l_name);
2693                 return 0;
2694         }
2695
2696         /* were we in a convert when we got the bast fire? */
2697         BUG_ON(lockres->l_action != OCFS2_AST_CONVERT &&
2698                lockres->l_action != OCFS2_AST_DOWNCONVERT);
2699         /* set things up for the unlockast to know to just
2700          * clear out the ast_action and unset busy, etc. */
2701         lockres->l_unlock_action = OCFS2_UNLOCK_CANCEL_CONVERT;
2702
2703         mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_BUSY),
2704                         "lock %s, invalid flags: 0x%lx\n",
2705                         lockres->l_name, lockres->l_flags);
2706
2707         return 1;
2708 }
2709
2710 static int ocfs2_cancel_convert(struct ocfs2_super *osb,
2711                                 struct ocfs2_lock_res *lockres)
2712 {
2713         int ret;
2714         enum dlm_status status;
2715
2716         mlog_entry_void();
2717         mlog(0, "lock %s\n", lockres->l_name);
2718
2719         ret = 0;
2720         status = dlmunlock(osb->dlm,
2721                            &lockres->l_lksb,
2722                            LKM_CANCEL,
2723                            ocfs2_unlock_ast,
2724                            lockres);
2725         if (status != DLM_NORMAL) {
2726                 ocfs2_log_dlm_error("dlmunlock", status, lockres);
2727                 ret = -EINVAL;
2728                 ocfs2_recover_from_dlm_error(lockres, 0);
2729         }
2730
2731         mlog(0, "lock %s return from dlmunlock\n", lockres->l_name);
2732
2733         mlog_exit(ret);
2734         return ret;
2735 }
2736
2737 static int ocfs2_unblock_lock(struct ocfs2_super *osb,
2738                               struct ocfs2_lock_res *lockres,
2739                               struct ocfs2_unblock_ctl *ctl)
2740 {
2741         unsigned long flags;
2742         int blocking;
2743         int new_level;
2744         int ret = 0;
2745         int set_lvb = 0;
2746
2747         mlog_entry_void();
2748
2749         spin_lock_irqsave(&lockres->l_lock, flags);
2750
2751         BUG_ON(!(lockres->l_flags & OCFS2_LOCK_BLOCKED));
2752
2753 recheck:
2754         if (lockres->l_flags & OCFS2_LOCK_BUSY) {
2755                 ctl->requeue = 1;
2756                 ret = ocfs2_prepare_cancel_convert(osb, lockres);
2757                 spin_unlock_irqrestore(&lockres->l_lock, flags);
2758                 if (ret) {
2759                         ret = ocfs2_cancel_convert(osb, lockres);
2760                         if (ret < 0)
2761                                 mlog_errno(ret);
2762                 }
2763                 goto leave;
2764         }
2765
2766         /* if we're blocking an exclusive and we have *any* holders,
2767          * then requeue. */
2768         if ((lockres->l_blocking == LKM_EXMODE)
2769             && (lockres->l_ex_holders || lockres->l_ro_holders))
2770                 goto leave_requeue;
2771
2772         /* If it's a PR we're blocking, then only
2773          * requeue if we've got any EX holders */
2774         if (lockres->l_blocking == LKM_PRMODE &&
2775             lockres->l_ex_holders)
2776                 goto leave_requeue;
2777
2778         /*
2779          * Can we get a lock in this state if the holder counts are
2780          * zero? The meta data unblock code used to check this.
2781          */
2782         if ((lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
2783             && (lockres->l_flags & OCFS2_LOCK_REFRESHING))
2784                 goto leave_requeue;
2785
2786         new_level = ocfs2_highest_compat_lock_level(lockres->l_blocking);
2787
2788         if (lockres->l_ops->check_downconvert
2789             && !lockres->l_ops->check_downconvert(lockres, new_level))
2790                 goto leave_requeue;
2791
2792         /* If we get here, then we know that there are no more
2793          * incompatible holders (and anyone asking for an incompatible
2794          * lock is blocked). We can now downconvert the lock */
2795         if (!lockres->l_ops->downconvert_worker)
2796                 goto downconvert;
2797
2798         /* Some lockres types want to do a bit of work before
2799          * downconverting a lock. Allow that here. The worker function
2800          * may sleep, so we save off a copy of what we're blocking as
2801          * it may change while we're not holding the spin lock. */
2802         blocking = lockres->l_blocking;
2803         spin_unlock_irqrestore(&lockres->l_lock, flags);
2804
2805         ctl->unblock_action = lockres->l_ops->downconvert_worker(lockres, blocking);
2806
2807         if (ctl->unblock_action == UNBLOCK_STOP_POST)
2808                 goto leave;
2809
2810         spin_lock_irqsave(&lockres->l_lock, flags);
2811         if (blocking != lockres->l_blocking) {
2812                 /* If this changed underneath us, then we can't drop
2813                  * it just yet. */
2814                 goto recheck;
2815         }
2816
2817 downconvert:
2818         ctl->requeue = 0;
2819
2820         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
2821                 if (lockres->l_level == LKM_EXMODE)
2822                         set_lvb = 1;
2823
2824                 /*
2825                  * We only set the lvb if the lock has been fully
2826                  * refreshed - otherwise we risk setting stale
2827                  * data. Otherwise, there's no need to actually clear
2828                  * out the lvb here as it's value is still valid.
2829                  */
2830                 if (set_lvb && !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
2831                         lockres->l_ops->set_lvb(lockres);
2832         }
2833
2834         ocfs2_prepare_downconvert(lockres, new_level);
2835         spin_unlock_irqrestore(&lockres->l_lock, flags);
2836         ret = ocfs2_downconvert_lock(osb, lockres, new_level, set_lvb);
2837 leave:
2838         mlog_exit(ret);
2839         return ret;
2840
2841 leave_requeue:
2842         spin_unlock_irqrestore(&lockres->l_lock, flags);
2843         ctl->requeue = 1;
2844
2845         mlog_exit(0);
2846         return 0;
2847 }
2848
2849 static int ocfs2_data_convert_worker(struct ocfs2_lock_res *lockres,
2850                                      int blocking)
2851 {
2852         struct inode *inode;
2853         struct address_space *mapping;
2854
2855         inode = ocfs2_lock_res_inode(lockres);
2856         mapping = inode->i_mapping;
2857
2858         /*
2859          * We need this before the filemap_fdatawrite() so that it can
2860          * transfer the dirty bit from the PTE to the
2861          * page. Unfortunately this means that even for EX->PR
2862          * downconverts, we'll lose our mappings and have to build
2863          * them up again.
2864          */
2865         unmap_mapping_range(mapping, 0, 0, 0);
2866
2867         if (filemap_fdatawrite(mapping)) {
2868                 mlog(ML_ERROR, "Could not sync inode %llu for downconvert!",
2869                      (unsigned long long)OCFS2_I(inode)->ip_blkno);
2870         }
2871         sync_mapping_buffers(mapping);
2872         if (blocking == LKM_EXMODE) {
2873                 truncate_inode_pages(mapping, 0);
2874         } else {
2875                 /* We only need to wait on the I/O if we're not also
2876                  * truncating pages because truncate_inode_pages waits
2877                  * for us above. We don't truncate pages if we're
2878                  * blocking anything < EXMODE because we want to keep
2879                  * them around in that case. */
2880                 filemap_fdatawait(mapping);
2881         }
2882
2883         return UNBLOCK_CONTINUE;
2884 }
2885
2886 static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
2887                                         int new_level)
2888 {
2889         struct inode *inode = ocfs2_lock_res_inode(lockres);
2890         int checkpointed = ocfs2_inode_fully_checkpointed(inode);
2891
2892         BUG_ON(new_level != LKM_NLMODE && new_level != LKM_PRMODE);
2893         BUG_ON(lockres->l_level != LKM_EXMODE && !checkpointed);
2894
2895         if (checkpointed)
2896                 return 1;
2897
2898         ocfs2_start_checkpoint(OCFS2_SB(inode->i_sb));
2899         return 0;
2900 }
2901
2902 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres)
2903 {
2904         struct inode *inode = ocfs2_lock_res_inode(lockres);
2905
2906         __ocfs2_stuff_meta_lvb(inode);
2907 }
2908
2909 /*
2910  * Does the final reference drop on our dentry lock. Right now this
2911  * happens in the vote thread, but we could choose to simplify the
2912  * dlmglue API and push these off to the ocfs2_wq in the future.
2913  */
2914 static void ocfs2_dentry_post_unlock(struct ocfs2_super *osb,
2915                                      struct ocfs2_lock_res *lockres)
2916 {
2917         struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
2918         ocfs2_dentry_lock_put(osb, dl);
2919 }
2920
2921 /*
2922  * d_delete() matching dentries before the lock downconvert.
2923  *
2924  * At this point, any process waiting to destroy the
2925  * dentry_lock due to last ref count is stopped by the
2926  * OCFS2_LOCK_QUEUED flag.
2927  *
2928  * We have two potential problems
2929  *
2930  * 1) If we do the last reference drop on our dentry_lock (via dput)
2931  *    we'll wind up in ocfs2_release_dentry_lock(), waiting on
2932  *    the downconvert to finish. Instead we take an elevated
2933  *    reference and push the drop until after we've completed our
2934  *    unblock processing.
2935  *
2936  * 2) There might be another process with a final reference,
2937  *    waiting on us to finish processing. If this is the case, we
2938  *    detect it and exit out - there's no more dentries anyway.
2939  */
2940 static int ocfs2_dentry_convert_worker(struct ocfs2_lock_res *lockres,
2941                                        int blocking)
2942 {
2943         struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
2944         struct ocfs2_inode_info *oi = OCFS2_I(dl->dl_inode);
2945         struct dentry *dentry;
2946         unsigned long flags;
2947         int extra_ref = 0;
2948
2949         /*
2950          * This node is blocking another node from getting a read
2951          * lock. This happens when we've renamed within a
2952          * directory. We've forced the other nodes to d_delete(), but
2953          * we never actually dropped our lock because it's still
2954          * valid. The downconvert code will retain a PR for this node,
2955          * so there's no further work to do.
2956          */
2957         if (blocking == LKM_PRMODE)
2958                 return UNBLOCK_CONTINUE;
2959
2960         /*
2961          * Mark this inode as potentially orphaned. The code in
2962          * ocfs2_delete_inode() will figure out whether it actually
2963          * needs to be freed or not.
2964          */
2965         spin_lock(&oi->ip_lock);
2966         oi->ip_flags |= OCFS2_INODE_MAYBE_ORPHANED;
2967         spin_unlock(&oi->ip_lock);
2968
2969         /*
2970          * Yuck. We need to make sure however that the check of
2971          * OCFS2_LOCK_FREEING and the extra reference are atomic with
2972          * respect to a reference decrement or the setting of that
2973          * flag.
2974          */
2975         spin_lock_irqsave(&lockres->l_lock, flags);
2976         spin_lock(&dentry_attach_lock);
2977         if (!(lockres->l_flags & OCFS2_LOCK_FREEING)
2978             && dl->dl_count) {
2979                 dl->dl_count++;
2980                 extra_ref = 1;
2981         }
2982         spin_unlock(&dentry_attach_lock);
2983         spin_unlock_irqrestore(&lockres->l_lock, flags);
2984
2985         mlog(0, "extra_ref = %d\n", extra_ref);
2986
2987         /*
2988          * We have a process waiting on us in ocfs2_dentry_iput(),
2989          * which means we can't have any more outstanding
2990          * aliases. There's no need to do any more work.
2991          */
2992         if (!extra_ref)
2993                 return UNBLOCK_CONTINUE;
2994
2995         spin_lock(&dentry_attach_lock);
2996         while (1) {
2997                 dentry = ocfs2_find_local_alias(dl->dl_inode,
2998                                                 dl->dl_parent_blkno, 1);
2999                 if (!dentry)
3000                         break;
3001                 spin_unlock(&dentry_attach_lock);
3002
3003                 mlog(0, "d_delete(%.*s);\n", dentry->d_name.len,
3004                      dentry->d_name.name);
3005
3006                 /*
3007                  * The following dcache calls may do an
3008                  * iput(). Normally we don't want that from the
3009                  * downconverting thread, but in this case it's ok
3010                  * because the requesting node already has an
3011                  * exclusive lock on the inode, so it can't be queued
3012                  * for a downconvert.
3013                  */
3014                 d_delete(dentry);
3015                 dput(dentry);
3016
3017                 spin_lock(&dentry_attach_lock);
3018         }
3019         spin_unlock(&dentry_attach_lock);
3020
3021         /*
3022          * If we are the last holder of this dentry lock, there is no
3023          * reason to downconvert so skip straight to the unlock.
3024          */
3025         if (dl->dl_count == 1)
3026                 return UNBLOCK_STOP_POST;
3027
3028         return UNBLOCK_CONTINUE_POST;
3029 }
3030
3031 void ocfs2_process_blocked_lock(struct ocfs2_super *osb,
3032                                 struct ocfs2_lock_res *lockres)
3033 {
3034         int status;
3035         struct ocfs2_unblock_ctl ctl = {0, 0,};
3036         unsigned long flags;
3037
3038         /* Our reference to the lockres in this function can be
3039          * considered valid until we remove the OCFS2_LOCK_QUEUED
3040          * flag. */
3041
3042         mlog_entry_void();
3043
3044         BUG_ON(!lockres);
3045         BUG_ON(!lockres->l_ops);
3046
3047         mlog(0, "lockres %s blocked.\n", lockres->l_name);
3048
3049         /* Detect whether a lock has been marked as going away while
3050          * the vote thread was processing other things. A lock can
3051          * still be marked with OCFS2_LOCK_FREEING after this check,
3052          * but short circuiting here will still save us some
3053          * performance. */
3054         spin_lock_irqsave(&lockres->l_lock, flags);
3055         if (lockres->l_flags & OCFS2_LOCK_FREEING)
3056                 goto unqueue;
3057         spin_unlock_irqrestore(&lockres->l_lock, flags);
3058
3059         status = ocfs2_unblock_lock(osb, lockres, &ctl);
3060         if (status < 0)
3061                 mlog_errno(status);
3062
3063         spin_lock_irqsave(&lockres->l_lock, flags);
3064 unqueue:
3065         if (lockres->l_flags & OCFS2_LOCK_FREEING || !ctl.requeue) {
3066                 lockres_clear_flags(lockres, OCFS2_LOCK_QUEUED);
3067         } else
3068                 ocfs2_schedule_blocked_lock(osb, lockres);
3069
3070         mlog(0, "lockres %s, requeue = %s.\n", lockres->l_name,
3071              ctl.requeue ? "yes" : "no");
3072         spin_unlock_irqrestore(&lockres->l_lock, flags);
3073
3074         if (ctl.unblock_action != UNBLOCK_CONTINUE
3075             && lockres->l_ops->post_unlock)
3076                 lockres->l_ops->post_unlock(osb, lockres);
3077
3078         mlog_exit_void();
3079 }
3080
3081 static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
3082                                         struct ocfs2_lock_res *lockres)
3083 {
3084         mlog_entry_void();
3085
3086         assert_spin_locked(&lockres->l_lock);
3087
3088         if (lockres->l_flags & OCFS2_LOCK_FREEING) {
3089                 /* Do not schedule a lock for downconvert when it's on
3090                  * the way to destruction - any nodes wanting access
3091                  * to the resource will get it soon. */
3092                 mlog(0, "Lockres %s won't be scheduled: flags 0x%lx\n",
3093                      lockres->l_name, lockres->l_flags);
3094                 return;
3095         }
3096
3097         lockres_or_flags(lockres, OCFS2_LOCK_QUEUED);
3098
3099         spin_lock(&osb->vote_task_lock);
3100         if (list_empty(&lockres->l_blocked_list)) {
3101                 list_add_tail(&lockres->l_blocked_list,
3102                               &osb->blocked_lock_list);
3103                 osb->blocked_lock_count++;
3104         }
3105         spin_unlock(&osb->vote_task_lock);
3106
3107         mlog_exit_void();
3108 }