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