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