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