root/fs/ocfs2/dlmglue.c

DEFINITIONS

1. ocfs2_dump_meta_lvb_info
2. ocfs2_is_inode_lock
3. ocfs2_lksb_to_lock_res
4. ocfs2_lock_res_inode
5. ocfs2_lock_res_dl
6. ocfs2_lock_res_qinfo
7. ocfs2_lock_res_refcount_tree
8. ocfs2_get_lockres_osb
9. ocfs2_cluster_unlock
10. ocfs2_build_lock_name
11. ocfs2_add_lockres_tracking
12. ocfs2_remove_lockres_tracking
13. ocfs2_init_lock_stats
14. ocfs2_update_lock_stats
15. ocfs2_track_lock_refresh
16. ocfs2_track_lock_wait
17. ocfs2_init_start_time
18. ocfs2_init_lock_stats
19. ocfs2_update_lock_stats
20. ocfs2_track_lock_refresh
21. ocfs2_track_lock_wait
22. ocfs2_init_start_time
23. ocfs2_lock_res_init_common
24. ocfs2_lock_res_init_once
25. ocfs2_inode_lock_res_init
26. ocfs2_get_inode_osb
27. ocfs2_get_qinfo_osb
28. ocfs2_get_file_osb
29. ocfs2_get_dentry_lock_ino
30. ocfs2_get_dentry_osb
31. ocfs2_dentry_lock_res_init
32. ocfs2_super_lock_res_init
33. ocfs2_rename_lock_res_init
34. ocfs2_nfs_sync_lock_res_init
35. ocfs2_trim_fs_lock_res_init
36. ocfs2_trim_fs_lock_res_uninit
37. ocfs2_orphan_scan_lock_res_init
38. ocfs2_file_lock_res_init
39. ocfs2_qinfo_lock_res_init
40. ocfs2_refcount_lock_res_init
41. ocfs2_lock_res_free
42. ocfs2_add_holder
43. ocfs2_pid_holder
44. ocfs2_remove_holder
45. ocfs2_inc_holders
46. ocfs2_dec_holders
47. ocfs2_highest_compat_lock_level
48. lockres_set_flags
49. lockres_or_flags
50. lockres_clear_flags
51. ocfs2_generic_handle_downconvert_action
52. ocfs2_generic_handle_convert_action
53. ocfs2_generic_handle_attach_action
54. ocfs2_generic_handle_bast
55. __lockres_clear_pending
56. lockres_clear_pending
57. lockres_set_pending
58. ocfs2_blocking_ast
59. ocfs2_locking_ast
60. ocfs2_unlock_ast
61. ocfs2_set_locking_protocol
62. ocfs2_recover_from_dlm_error
63. ocfs2_lock_create
64. ocfs2_check_wait_flag
65. ocfs2_wait_on_busy_lock
66. ocfs2_wait_on_refreshing_lock
67. ocfs2_may_continue_on_blocked_lock
68. ocfs2_init_mask_waiter
69. ocfs2_wait_for_mask
70. lockres_add_mask_waiter
71. __lockres_remove_mask_waiter
72. lockres_remove_mask_waiter
73. ocfs2_wait_for_mask_interruptible
74. __ocfs2_cluster_lock
75. ocfs2_cluster_lock
76. __ocfs2_cluster_unlock
77. ocfs2_create_new_lock
78. ocfs2_create_new_inode_locks
79. ocfs2_rw_lock
80. ocfs2_try_rw_lock
81. ocfs2_rw_unlock
82. ocfs2_open_lock
83. ocfs2_try_open_lock
84. ocfs2_open_unlock
85. ocfs2_flock_handle_signal
86. ocfs2_file_lock
87. ocfs2_file_unlock
88. ocfs2_downconvert_on_unlock
89. ocfs2_pack_timespec
90. __ocfs2_stuff_meta_lvb
91. ocfs2_unpack_timespec
92. ocfs2_refresh_inode_from_lvb
93. ocfs2_meta_lvb_is_trustable
94. ocfs2_should_refresh_lock_res
95. ocfs2_complete_lock_res_refresh
96. ocfs2_inode_lock_update
97. ocfs2_assign_bh
98. ocfs2_inode_lock_full_nested
99. ocfs2_inode_lock_with_page
100. ocfs2_inode_lock_atime
101. ocfs2_inode_unlock
102. ocfs2_inode_lock_tracker
103. ocfs2_inode_unlock_tracker
104. ocfs2_orphan_scan_lock
105. ocfs2_orphan_scan_unlock
106. ocfs2_super_lock
107. ocfs2_super_unlock
108. ocfs2_rename_lock
109. ocfs2_rename_unlock
110. ocfs2_nfs_sync_lock
111. ocfs2_nfs_sync_unlock
112. ocfs2_trim_fs_lock
113. ocfs2_trim_fs_unlock
114. ocfs2_dentry_lock
115. ocfs2_dentry_unlock
116. ocfs2_dlm_debug_free
117. ocfs2_put_dlm_debug
118. ocfs2_get_dlm_debug
119. ocfs2_new_dlm_debug
120. ocfs2_dlm_next_res
121. ocfs2_dlm_seq_start
122. ocfs2_dlm_seq_stop
123. ocfs2_dlm_seq_next
124. ocfs2_dlm_seq_show
125. ocfs2_dlm_debug_release
126. ocfs2_dlm_debug_open
127. ocfs2_dlm_init_debug
128. ocfs2_dlm_shutdown_debug
129. ocfs2_dlm_init
130. ocfs2_dlm_shutdown
131. ocfs2_drop_lock
132. ocfs2_mark_lockres_freeing
133. ocfs2_simple_drop_lockres
134. ocfs2_drop_osb_locks
135. ocfs2_drop_inode_locks
136. ocfs2_prepare_downconvert
137. ocfs2_downconvert_lock
138. ocfs2_prepare_cancel_convert
139. ocfs2_cancel_convert
140. ocfs2_unblock_lock
141. ocfs2_data_convert_worker
142. ocfs2_ci_checkpointed
143. ocfs2_check_meta_downconvert
144. ocfs2_set_meta_lvb
145. ocfs2_dentry_post_unlock
146. ocfs2_dentry_convert_worker
147. ocfs2_check_refcount_downconvert
148. ocfs2_refcount_convert_worker
149. ocfs2_set_qinfo_lvb
150. ocfs2_qinfo_unlock
151. ocfs2_refresh_qinfo
152. ocfs2_qinfo_lock
153. ocfs2_refcount_lock
154. ocfs2_refcount_unlock
155. ocfs2_process_blocked_lock
156. ocfs2_schedule_blocked_lock
157. ocfs2_downconvert_thread_do_work
158. ocfs2_downconvert_thread_lists_empty
159. ocfs2_downconvert_thread_should_wake
160. ocfs2_downconvert_thread
161. ocfs2_wake_downconvert_thread

   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         ocfs2_get_dlm_debug(dlm_debug);
3286 }
3287 
3288 static void ocfs2_dlm_shutdown_debug(struct ocfs2_super *osb)
3289 {
3290         struct ocfs2_dlm_debug *dlm_debug = osb->osb_dlm_debug;
3291 
3292         if (dlm_debug)
3293                 ocfs2_put_dlm_debug(dlm_debug);
3294 }
3295 
3296 int ocfs2_dlm_init(struct ocfs2_super *osb)
3297 {
3298         int status = 0;
3299         struct ocfs2_cluster_connection *conn = NULL;
3300 
3301         if (ocfs2_mount_local(osb)) {
3302                 osb->node_num = 0;
3303                 goto local;
3304         }
3305 
3306         ocfs2_dlm_init_debug(osb);
3307 
3308         /* launch downconvert thread */
3309         osb->dc_task = kthread_run(ocfs2_downconvert_thread, osb, "ocfs2dc-%s",
3310                         osb->uuid_str);
3311         if (IS_ERR(osb->dc_task)) {
3312                 status = PTR_ERR(osb->dc_task);
3313                 osb->dc_task = NULL;
3314                 mlog_errno(status);
3315                 goto bail;
3316         }
3317 
3318         /* for now, uuid == domain */
3319         status = ocfs2_cluster_connect(osb->osb_cluster_stack,
3320                                        osb->osb_cluster_name,
3321                                        strlen(osb->osb_cluster_name),
3322                                        osb->uuid_str,
3323                                        strlen(osb->uuid_str),
3324                                        &lproto, ocfs2_do_node_down, osb,
3325                                        &conn);
3326         if (status) {
3327                 mlog_errno(status);
3328                 goto bail;
3329         }
3330 
3331         status = ocfs2_cluster_this_node(conn, &osb->node_num);
3332         if (status < 0) {
3333                 mlog_errno(status);
3334                 mlog(ML_ERROR,
3335                      "could not find this host's node number\n");
3336                 ocfs2_cluster_disconnect(conn, 0);
3337                 goto bail;
3338         }
3339 
3340 local:
3341         ocfs2_super_lock_res_init(&osb->osb_super_lockres, osb);
3342         ocfs2_rename_lock_res_init(&osb->osb_rename_lockres, osb);
3343         ocfs2_nfs_sync_lock_res_init(&osb->osb_nfs_sync_lockres, osb);
3344         ocfs2_orphan_scan_lock_res_init(&osb->osb_orphan_scan.os_lockres, osb);
3345 
3346         osb->cconn = conn;
3347 bail:
3348         if (status < 0) {
3349                 ocfs2_dlm_shutdown_debug(osb);
3350                 if (osb->dc_task)
3351                         kthread_stop(osb->dc_task);
3352         }
3353 
3354         return status;
3355 }
3356 
3357 void ocfs2_dlm_shutdown(struct ocfs2_super *osb,
3358                         int hangup_pending)
3359 {
3360         ocfs2_drop_osb_locks(osb);
3361 
3362         /*
3363          * Now that we have dropped all locks and ocfs2_dismount_volume()
3364          * has disabled recovery, the DLM won't be talking to us.  It's
3365          * safe to tear things down before disconnecting the cluster.
3366          */
3367 
3368         if (osb->dc_task) {
3369                 kthread_stop(osb->dc_task);
3370                 osb->dc_task = NULL;
3371         }
3372 
3373         ocfs2_lock_res_free(&osb->osb_super_lockres);
3374         ocfs2_lock_res_free(&osb->osb_rename_lockres);
3375         ocfs2_lock_res_free(&osb->osb_nfs_sync_lockres);
3376         ocfs2_lock_res_free(&osb->osb_orphan_scan.os_lockres);
3377 
3378         ocfs2_cluster_disconnect(osb->cconn, hangup_pending);
3379         osb->cconn = NULL;
3380 
3381         ocfs2_dlm_shutdown_debug(osb);
3382 }
3383 
3384 static int ocfs2_drop_lock(struct ocfs2_super *osb,
3385                            struct ocfs2_lock_res *lockres)
3386 {
3387         int ret;
3388         unsigned long flags;
3389         u32 lkm_flags = 0;
3390 
3391         /* We didn't get anywhere near actually using this lockres. */
3392         if (!(lockres->l_flags & OCFS2_LOCK_INITIALIZED))
3393                 goto out;
3394 
3395         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
3396                 lkm_flags |= DLM_LKF_VALBLK;
3397 
3398         spin_lock_irqsave(&lockres->l_lock, flags);
3399 
3400         mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_FREEING),
3401                         "lockres %s, flags 0x%lx\n",
3402                         lockres->l_name, lockres->l_flags);
3403 
3404         while (lockres->l_flags & OCFS2_LOCK_BUSY) {
3405                 mlog(0, "waiting on busy lock \"%s\": flags = %lx, action = "
3406                      "%u, unlock_action = %u\n",
3407                      lockres->l_name, lockres->l_flags, lockres->l_action,
3408                      lockres->l_unlock_action);
3409 
3410                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3411 
3412                 /* XXX: Today we just wait on any busy
3413                  * locks... Perhaps we need to cancel converts in the
3414                  * future? */
3415                 ocfs2_wait_on_busy_lock(lockres);
3416 
3417                 spin_lock_irqsave(&lockres->l_lock, flags);
3418         }
3419 
3420         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
3421                 if (lockres->l_flags & OCFS2_LOCK_ATTACHED &&
3422                     lockres->l_level == DLM_LOCK_EX &&
3423                     !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
3424                         lockres->l_ops->set_lvb(lockres);
3425         }
3426 
3427         if (lockres->l_flags & OCFS2_LOCK_BUSY)
3428                 mlog(ML_ERROR, "destroying busy lock: \"%s\"\n",
3429                      lockres->l_name);
3430         if (lockres->l_flags & OCFS2_LOCK_BLOCKED)
3431                 mlog(0, "destroying blocked lock: \"%s\"\n", lockres->l_name);
3432 
3433         if (!(lockres->l_flags & OCFS2_LOCK_ATTACHED)) {
3434                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3435                 goto out;
3436         }
3437 
3438         lockres_clear_flags(lockres, OCFS2_LOCK_ATTACHED);
3439 
3440         /* make sure we never get here while waiting for an ast to
3441          * fire. */
3442         BUG_ON(lockres->l_action != OCFS2_AST_INVALID);
3443 
3444         /* is this necessary? */
3445         lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
3446         lockres->l_unlock_action = OCFS2_UNLOCK_DROP_LOCK;
3447         spin_unlock_irqrestore(&lockres->l_lock, flags);
3448 
3449         mlog(0, "lock %s\n", lockres->l_name);
3450 
3451         ret = ocfs2_dlm_unlock(osb->cconn, &lockres->l_lksb, lkm_flags);
3452         if (ret) {
3453                 ocfs2_log_dlm_error("ocfs2_dlm_unlock", ret, lockres);
3454                 mlog(ML_ERROR, "lockres flags: %lu\n", lockres->l_flags);
3455                 ocfs2_dlm_dump_lksb(&lockres->l_lksb);
3456                 BUG();
3457         }
3458         mlog(0, "lock %s, successful return from ocfs2_dlm_unlock\n",
3459              lockres->l_name);
3460 
3461         ocfs2_wait_on_busy_lock(lockres);
3462 out:
3463         return 0;
3464 }
3465 
3466 static void ocfs2_process_blocked_lock(struct ocfs2_super *osb,
3467                                        struct ocfs2_lock_res *lockres);
3468 
3469 /* Mark the lockres as being dropped. It will no longer be
3470  * queued if blocking, but we still may have to wait on it
3471  * being dequeued from the downconvert thread before we can consider
3472  * it safe to drop.
3473  *
3474  * You can *not* attempt to call cluster_lock on this lockres anymore. */
3475 void ocfs2_mark_lockres_freeing(struct ocfs2_super *osb,
3476                                 struct ocfs2_lock_res *lockres)
3477 {
3478         int status;
3479         struct ocfs2_mask_waiter mw;
3480         unsigned long flags, flags2;
3481 
3482         ocfs2_init_mask_waiter(&mw);
3483 
3484         spin_lock_irqsave(&lockres->l_lock, flags);
3485         lockres->l_flags |= OCFS2_LOCK_FREEING;
3486         if (lockres->l_flags & OCFS2_LOCK_QUEUED && current == osb->dc_task) {
3487                 /*
3488                  * We know the downconvert is queued but not in progress
3489                  * because we are the downconvert thread and processing
3490                  * different lock. So we can just remove the lock from the
3491                  * queue. This is not only an optimization but also a way
3492                  * to avoid the following deadlock:
3493                  *   ocfs2_dentry_post_unlock()
3494                  *     ocfs2_dentry_lock_put()
3495                  *       ocfs2_drop_dentry_lock()
3496                  *         iput()
3497                  *           ocfs2_evict_inode()
3498                  *             ocfs2_clear_inode()
3499                  *               ocfs2_mark_lockres_freeing()
3500                  *                 ... blocks waiting for OCFS2_LOCK_QUEUED
3501                  *                 since we are the downconvert thread which
3502                  *                 should clear the flag.
3503                  */
3504                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3505                 spin_lock_irqsave(&osb->dc_task_lock, flags2);
3506                 list_del_init(&lockres->l_blocked_list);
3507                 osb->blocked_lock_count--;
3508                 spin_unlock_irqrestore(&osb->dc_task_lock, flags2);
3509                 /*
3510                  * Warn if we recurse into another post_unlock call.  Strictly
3511                  * speaking it isn't a problem but we need to be careful if
3512                  * that happens (stack overflow, deadlocks, ...) so warn if
3513                  * ocfs2 grows a path for which this can happen.
3514                  */
3515                 WARN_ON_ONCE(lockres->l_ops->post_unlock);
3516                 /* Since the lock is freeing we don't do much in the fn below */
3517                 ocfs2_process_blocked_lock(osb, lockres);
3518                 return;
3519         }
3520         while (lockres->l_flags & OCFS2_LOCK_QUEUED) {
3521                 lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_QUEUED, 0);
3522                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3523 
3524                 mlog(0, "Waiting on lockres %s\n", lockres->l_name);
3525 
3526                 status = ocfs2_wait_for_mask(&mw);
3527                 if (status)
3528                         mlog_errno(status);
3529 
3530                 spin_lock_irqsave(&lockres->l_lock, flags);
3531         }
3532         spin_unlock_irqrestore(&lockres->l_lock, flags);
3533 }
3534 
3535 void ocfs2_simple_drop_lockres(struct ocfs2_super *osb,
3536                                struct ocfs2_lock_res *lockres)
3537 {
3538         int ret;
3539 
3540         ocfs2_mark_lockres_freeing(osb, lockres);
3541         ret = ocfs2_drop_lock(osb, lockres);
3542         if (ret)
3543                 mlog_errno(ret);
3544 }
3545 
3546 static void ocfs2_drop_osb_locks(struct ocfs2_super *osb)
3547 {
3548         ocfs2_simple_drop_lockres(osb, &osb->osb_super_lockres);
3549         ocfs2_simple_drop_lockres(osb, &osb->osb_rename_lockres);
3550         ocfs2_simple_drop_lockres(osb, &osb->osb_nfs_sync_lockres);
3551         ocfs2_simple_drop_lockres(osb, &osb->osb_orphan_scan.os_lockres);
3552 }
3553 
3554 int ocfs2_drop_inode_locks(struct inode *inode)
3555 {
3556         int status, err;
3557 
3558         /* No need to call ocfs2_mark_lockres_freeing here -
3559          * ocfs2_clear_inode has done it for us. */
3560 
3561         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3562                               &OCFS2_I(inode)->ip_open_lockres);
3563         if (err < 0)
3564                 mlog_errno(err);
3565 
3566         status = err;
3567 
3568         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3569                               &OCFS2_I(inode)->ip_inode_lockres);
3570         if (err < 0)
3571                 mlog_errno(err);
3572         if (err < 0 && !status)
3573                 status = err;
3574 
3575         err = ocfs2_drop_lock(OCFS2_SB(inode->i_sb),
3576                               &OCFS2_I(inode)->ip_rw_lockres);
3577         if (err < 0)
3578                 mlog_errno(err);
3579         if (err < 0 && !status)
3580                 status = err;
3581 
3582         return status;
3583 }
3584 
3585 static unsigned int ocfs2_prepare_downconvert(struct ocfs2_lock_res *lockres,
3586                                               int new_level)
3587 {
3588         assert_spin_locked(&lockres->l_lock);
3589 
3590         BUG_ON(lockres->l_blocking <= DLM_LOCK_NL);
3591 
3592         if (lockres->l_level <= new_level) {
3593                 mlog(ML_ERROR, "lockres %s, lvl %d <= %d, blcklst %d, mask %d, "
3594                      "type %d, flags 0x%lx, hold %d %d, act %d %d, req %d, "
3595                      "block %d, pgen %d\n", lockres->l_name, lockres->l_level,
3596                      new_level, list_empty(&lockres->l_blocked_list),
3597                      list_empty(&lockres->l_mask_waiters), lockres->l_type,
3598                      lockres->l_flags, lockres->l_ro_holders,
3599                      lockres->l_ex_holders, lockres->l_action,
3600                      lockres->l_unlock_action, lockres->l_requested,
3601                      lockres->l_blocking, lockres->l_pending_gen);
3602                 BUG();
3603         }
3604 
3605         mlog(ML_BASTS, "lockres %s, level %d => %d, blocking %d\n",
3606              lockres->l_name, lockres->l_level, new_level, lockres->l_blocking);
3607 
3608         lockres->l_action = OCFS2_AST_DOWNCONVERT;
3609         lockres->l_requested = new_level;
3610         lockres_or_flags(lockres, OCFS2_LOCK_BUSY);
3611         return lockres_set_pending(lockres);
3612 }
3613 
3614 static int ocfs2_downconvert_lock(struct ocfs2_super *osb,
3615                                   struct ocfs2_lock_res *lockres,
3616                                   int new_level,
3617                                   int lvb,
3618                                   unsigned int generation)
3619 {
3620         int ret;
3621         u32 dlm_flags = DLM_LKF_CONVERT;
3622 
3623         mlog(ML_BASTS, "lockres %s, level %d => %d\n", lockres->l_name,
3624              lockres->l_level, new_level);
3625 
3626         /*
3627          * On DLM_LKF_VALBLK, fsdlm behaves differently with o2cb. It always
3628          * expects DLM_LKF_VALBLK being set if the LKB has LVB, so that
3629          * we can recover correctly from node failure. Otherwise, we may get
3630          * invalid LVB in LKB, but without DLM_SBF_VALNOTVALID being set.
3631          */
3632         if (ocfs2_userspace_stack(osb) &&
3633             lockres->l_ops->flags & LOCK_TYPE_USES_LVB)
3634                 lvb = 1;
3635 
3636         if (lvb)
3637                 dlm_flags |= DLM_LKF_VALBLK;
3638 
3639         ret = ocfs2_dlm_lock(osb->cconn,
3640                              new_level,
3641                              &lockres->l_lksb,
3642                              dlm_flags,
3643                              lockres->l_name,
3644                              OCFS2_LOCK_ID_MAX_LEN - 1);
3645         lockres_clear_pending(lockres, generation, osb);
3646         if (ret) {
3647                 ocfs2_log_dlm_error("ocfs2_dlm_lock", ret, lockres);
3648                 ocfs2_recover_from_dlm_error(lockres, 1);
3649                 goto bail;
3650         }
3651 
3652         ret = 0;
3653 bail:
3654         return ret;
3655 }
3656 
3657 /* returns 1 when the caller should unlock and call ocfs2_dlm_unlock */
3658 static int ocfs2_prepare_cancel_convert(struct ocfs2_super *osb,
3659                                         struct ocfs2_lock_res *lockres)
3660 {
3661         assert_spin_locked(&lockres->l_lock);
3662 
3663         if (lockres->l_unlock_action == OCFS2_UNLOCK_CANCEL_CONVERT) {
3664                 /* If we're already trying to cancel a lock conversion
3665                  * then just drop the spinlock and allow the caller to
3666                  * requeue this lock. */
3667                 mlog(ML_BASTS, "lockres %s, skip convert\n", lockres->l_name);
3668                 return 0;
3669         }
3670 
3671         /* were we in a convert when we got the bast fire? */
3672         BUG_ON(lockres->l_action != OCFS2_AST_CONVERT &&
3673                lockres->l_action != OCFS2_AST_DOWNCONVERT);
3674         /* set things up for the unlockast to know to just
3675          * clear out the ast_action and unset busy, etc. */
3676         lockres->l_unlock_action = OCFS2_UNLOCK_CANCEL_CONVERT;
3677 
3678         mlog_bug_on_msg(!(lockres->l_flags & OCFS2_LOCK_BUSY),
3679                         "lock %s, invalid flags: 0x%lx\n",
3680                         lockres->l_name, lockres->l_flags);
3681 
3682         mlog(ML_BASTS, "lockres %s\n", lockres->l_name);
3683 
3684         return 1;
3685 }
3686 
3687 static int ocfs2_cancel_convert(struct ocfs2_super *osb,
3688                                 struct ocfs2_lock_res *lockres)
3689 {
3690         int ret;
3691 
3692         ret = ocfs2_dlm_unlock(osb->cconn, &lockres->l_lksb,
3693                                DLM_LKF_CANCEL);
3694         if (ret) {
3695                 ocfs2_log_dlm_error("ocfs2_dlm_unlock", ret, lockres);
3696                 ocfs2_recover_from_dlm_error(lockres, 0);
3697         }
3698 
3699         mlog(ML_BASTS, "lockres %s\n", lockres->l_name);
3700 
3701         return ret;
3702 }
3703 
3704 static int ocfs2_unblock_lock(struct ocfs2_super *osb,
3705                               struct ocfs2_lock_res *lockres,
3706                               struct ocfs2_unblock_ctl *ctl)
3707 {
3708         unsigned long flags;
3709         int blocking;
3710         int new_level;
3711         int level;
3712         int ret = 0;
3713         int set_lvb = 0;
3714         unsigned int gen;
3715 
3716         spin_lock_irqsave(&lockres->l_lock, flags);
3717 
3718 recheck:
3719         /*
3720          * Is it still blocking? If not, we have no more work to do.
3721          */
3722         if (!(lockres->l_flags & OCFS2_LOCK_BLOCKED)) {
3723                 BUG_ON(lockres->l_blocking != DLM_LOCK_NL);
3724                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3725                 ret = 0;
3726                 goto leave;
3727         }
3728 
3729         if (lockres->l_flags & OCFS2_LOCK_BUSY) {
3730                 /* XXX
3731                  * This is a *big* race.  The OCFS2_LOCK_PENDING flag
3732                  * exists entirely for one reason - another thread has set
3733                  * OCFS2_LOCK_BUSY, but has *NOT* yet called dlm_lock().
3734                  *
3735                  * If we do ocfs2_cancel_convert() before the other thread
3736                  * calls dlm_lock(), our cancel will do nothing.  We will
3737                  * get no ast, and we will have no way of knowing the
3738                  * cancel failed.  Meanwhile, the other thread will call
3739                  * into dlm_lock() and wait...forever.
3740                  *
3741                  * Why forever?  Because another node has asked for the
3742                  * lock first; that's why we're here in unblock_lock().
3743                  *
3744                  * The solution is OCFS2_LOCK_PENDING.  When PENDING is
3745                  * set, we just requeue the unblock.  Only when the other
3746                  * thread has called dlm_lock() and cleared PENDING will
3747                  * we then cancel their request.
3748                  *
3749                  * All callers of dlm_lock() must set OCFS2_DLM_PENDING
3750                  * at the same time they set OCFS2_DLM_BUSY.  They must
3751                  * clear OCFS2_DLM_PENDING after dlm_lock() returns.
3752                  */
3753                 if (lockres->l_flags & OCFS2_LOCK_PENDING) {
3754                         mlog(ML_BASTS, "lockres %s, ReQ: Pending\n",
3755                              lockres->l_name);
3756                         goto leave_requeue;
3757                 }
3758 
3759                 ctl->requeue = 1;
3760                 ret = ocfs2_prepare_cancel_convert(osb, lockres);
3761                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3762                 if (ret) {
3763                         ret = ocfs2_cancel_convert(osb, lockres);
3764                         if (ret < 0)
3765                                 mlog_errno(ret);
3766                 }
3767                 goto leave;
3768         }
3769 
3770         /*
3771          * This prevents livelocks. OCFS2_LOCK_UPCONVERT_FINISHING flag is
3772          * set when the ast is received for an upconvert just before the
3773          * OCFS2_LOCK_BUSY flag is cleared. Now if the fs received a bast
3774          * on the heels of the ast, we want to delay the downconvert just
3775          * enough to allow the up requestor to do its task. Because this
3776          * lock is in the blocked queue, the lock will be downconverted
3777          * as soon as the requestor is done with the lock.
3778          */
3779         if (lockres->l_flags & OCFS2_LOCK_UPCONVERT_FINISHING)
3780                 goto leave_requeue;
3781 
3782         /*
3783          * How can we block and yet be at NL?  We were trying to upconvert
3784          * from NL and got canceled.  The code comes back here, and now
3785          * we notice and clear BLOCKING.
3786          */
3787         if (lockres->l_level == DLM_LOCK_NL) {
3788                 BUG_ON(lockres->l_ex_holders || lockres->l_ro_holders);
3789                 mlog(ML_BASTS, "lockres %s, Aborting dc\n", lockres->l_name);
3790                 lockres->l_blocking = DLM_LOCK_NL;
3791                 lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED);
3792                 spin_unlock_irqrestore(&lockres->l_lock, flags);
3793                 goto leave;
3794         }
3795 
3796         /* if we're blocking an exclusive and we have *any* holders,
3797          * then requeue. */
3798         if ((lockres->l_blocking == DLM_LOCK_EX)
3799             && (lockres->l_ex_holders || lockres->l_ro_holders)) {
3800                 mlog(ML_BASTS, "lockres %s, ReQ: EX/PR Holders %u,%u\n",
3801                      lockres->l_name, lockres->l_ex_holders,
3802                      lockres->l_ro_holders);
3803                 goto leave_requeue;
3804         }
3805 
3806         /* If it's a PR we're blocking, then only
3807          * requeue if we've got any EX holders */
3808         if (lockres->l_blocking == DLM_LOCK_PR &&
3809             lockres->l_ex_holders) {
3810                 mlog(ML_BASTS, "lockres %s, ReQ: EX Holders %u\n",
3811                      lockres->l_name, lockres->l_ex_holders);
3812                 goto leave_requeue;
3813         }
3814 
3815         /*
3816          * Can we get a lock in this state if the holder counts are
3817          * zero? The meta data unblock code used to check this.
3818          */
3819         if ((lockres->l_ops->flags & LOCK_TYPE_REQUIRES_REFRESH)
3820             && (lockres->l_flags & OCFS2_LOCK_REFRESHING)) {
3821                 mlog(ML_BASTS, "lockres %s, ReQ: Lock Refreshing\n",
3822                      lockres->l_name);
3823                 goto leave_requeue;
3824         }
3825 
3826         new_level = ocfs2_highest_compat_lock_level(lockres->l_blocking);
3827 
3828         if (lockres->l_ops->check_downconvert
3829             && !lockres->l_ops->check_downconvert(lockres, new_level)) {
3830                 mlog(ML_BASTS, "lockres %s, ReQ: Checkpointing\n",
3831                      lockres->l_name);
3832                 goto leave_requeue;
3833         }
3834 
3835         /* If we get here, then we know that there are no more
3836          * incompatible holders (and anyone asking for an incompatible
3837          * lock is blocked). We can now downconvert the lock */
3838         if (!lockres->l_ops->downconvert_worker)
3839                 goto downconvert;
3840 
3841         /* Some lockres types want to do a bit of work before
3842          * downconverting a lock. Allow that here. The worker function
3843          * may sleep, so we save off a copy of what we're blocking as
3844          * it may change while we're not holding the spin lock. */
3845         blocking = lockres->l_blocking;
3846         level = lockres->l_level;
3847         spin_unlock_irqrestore(&lockres->l_lock, flags);
3848 
3849         ctl->unblock_action = lockres->l_ops->downconvert_worker(lockres, blocking);
3850 
3851         if (ctl->unblock_action == UNBLOCK_STOP_POST) {
3852                 mlog(ML_BASTS, "lockres %s, UNBLOCK_STOP_POST\n",
3853                      lockres->l_name);
3854                 goto leave;
3855         }
3856 
3857         spin_lock_irqsave(&lockres->l_lock, flags);
3858         if ((blocking != lockres->l_blocking) || (level != lockres->l_level)) {
3859                 /* If this changed underneath us, then we can't drop
3860                  * it just yet. */
3861                 mlog(ML_BASTS, "lockres %s, block=%d:%d, level=%d:%d, "
3862                      "Recheck\n", lockres->l_name, blocking,
3863                      lockres->l_blocking, level, lockres->l_level);
3864                 goto recheck;
3865         }
3866 
3867 downconvert:
3868         ctl->requeue = 0;
3869 
3870         if (lockres->l_ops->flags & LOCK_TYPE_USES_LVB) {
3871                 if (lockres->l_level == DLM_LOCK_EX)
3872                         set_lvb = 1;
3873 
3874                 /*
3875                  * We only set the lvb if the lock has been fully
3876                  * refreshed - otherwise we risk setting stale
3877                  * data. Otherwise, there's no need to actually clear
3878                  * out the lvb here as it's value is still valid.
3879                  */
3880                 if (set_lvb && !(lockres->l_flags & OCFS2_LOCK_NEEDS_REFRESH))
3881                         lockres->l_ops->set_lvb(lockres);
3882         }
3883 
3884         gen = ocfs2_prepare_downconvert(lockres, new_level);
3885         spin_unlock_irqrestore(&lockres->l_lock, flags);
3886         ret = ocfs2_downconvert_lock(osb, lockres, new_level, set_lvb,
3887                                      gen);
3888 
3889 leave:
3890         if (ret)
3891                 mlog_errno(ret);
3892         return ret;
3893 
3894 leave_requeue:
3895         spin_unlock_irqrestore(&lockres->l_lock, flags);
3896         ctl->requeue = 1;
3897 
3898         return 0;
3899 }
3900 
3901 static int ocfs2_data_convert_worker(struct ocfs2_lock_res *lockres,
3902                                      int blocking)
3903 {
3904         struct inode *inode;
3905         struct address_space *mapping;
3906         struct ocfs2_inode_info *oi;
3907 
3908         inode = ocfs2_lock_res_inode(lockres);
3909         mapping = inode->i_mapping;
3910 
3911         if (S_ISDIR(inode->i_mode)) {
3912                 oi = OCFS2_I(inode);
3913                 oi->ip_dir_lock_gen++;
3914                 mlog(0, "generation: %u\n", oi->ip_dir_lock_gen);
3915                 goto out;
3916         }
3917 
3918         if (!S_ISREG(inode->i_mode))
3919                 goto out;
3920 
3921         /*
3922          * We need this before the filemap_fdatawrite() so that it can
3923          * transfer the dirty bit from the PTE to the
3924          * page. Unfortunately this means that even for EX->PR
3925          * downconverts, we'll lose our mappings and have to build
3926          * them up again.
3927          */
3928         unmap_mapping_range(mapping, 0, 0, 0);
3929 
3930         if (filemap_fdatawrite(mapping)) {
3931                 mlog(ML_ERROR, "Could not sync inode %llu for downconvert!",
3932                      (unsigned long long)OCFS2_I(inode)->ip_blkno);
3933         }
3934         sync_mapping_buffers(mapping);
3935         if (blocking == DLM_LOCK_EX) {
3936                 truncate_inode_pages(mapping, 0);
3937         } else {
3938                 /* We only need to wait on the I/O if we're not also
3939                  * truncating pages because truncate_inode_pages waits
3940                  * for us above. We don't truncate pages if we're
3941                  * blocking anything < EXMODE because we want to keep
3942                  * them around in that case. */
3943                 filemap_fdatawait(mapping);
3944         }
3945 
3946         forget_all_cached_acls(inode);
3947 
3948 out:
3949         return UNBLOCK_CONTINUE;
3950 }
3951 
3952 static int ocfs2_ci_checkpointed(struct ocfs2_caching_info *ci,
3953                                  struct ocfs2_lock_res *lockres,
3954                                  int new_level)
3955 {
3956         int checkpointed = ocfs2_ci_fully_checkpointed(ci);
3957 
3958         BUG_ON(new_level != DLM_LOCK_NL && new_level != DLM_LOCK_PR);
3959         BUG_ON(lockres->l_level != DLM_LOCK_EX && !checkpointed);
3960 
3961         if (checkpointed)
3962                 return 1;
3963 
3964         ocfs2_start_checkpoint(OCFS2_SB(ocfs2_metadata_cache_get_super(ci)));
3965         return 0;
3966 }
3967 
3968 static int ocfs2_check_meta_downconvert(struct ocfs2_lock_res *lockres,
3969                                         int new_level)
3970 {
3971         struct inode *inode = ocfs2_lock_res_inode(lockres);
3972 
3973         return ocfs2_ci_checkpointed(INODE_CACHE(inode), lockres, new_level);
3974 }
3975 
3976 static void ocfs2_set_meta_lvb(struct ocfs2_lock_res *lockres)
3977 {
3978         struct inode *inode = ocfs2_lock_res_inode(lockres);
3979 
3980         __ocfs2_stuff_meta_lvb(inode);
3981 }
3982 
3983 /*
3984  * Does the final reference drop on our dentry lock. Right now this
3985  * happens in the downconvert thread, but we could choose to simplify the
3986  * dlmglue API and push these off to the ocfs2_wq in the future.
3987  */
3988 static void ocfs2_dentry_post_unlock(struct ocfs2_super *osb,
3989                                      struct ocfs2_lock_res *lockres)
3990 {
3991         struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
3992         ocfs2_dentry_lock_put(osb, dl);
3993 }
3994 
3995 /*
3996  * d_delete() matching dentries before the lock downconvert.
3997  *
3998  * At this point, any process waiting to destroy the
3999  * dentry_lock due to last ref count is stopped by the
4000  * OCFS2_LOCK_QUEUED flag.
4001  *
4002  * We have two potential problems
4003  *
4004  * 1) If we do the last reference drop on our dentry_lock (via dput)
4005  *    we'll wind up in ocfs2_release_dentry_lock(), waiting on
4006  *    the downconvert to finish. Instead we take an elevated
4007  *    reference and push the drop until after we've completed our
4008  *    unblock processing.
4009  *
4010  * 2) There might be another process with a final reference,
4011  *    waiting on us to finish processing. If this is the case, we
4012  *    detect it and exit out - there's no more dentries anyway.
4013  */
4014 static int ocfs2_dentry_convert_worker(struct ocfs2_lock_res *lockres,
4015                                        int blocking)
4016 {
4017         struct ocfs2_dentry_lock *dl = ocfs2_lock_res_dl(lockres);
4018         struct ocfs2_inode_info *oi = OCFS2_I(dl->dl_inode);
4019         struct dentry *dentry;
4020         unsigned long flags;
4021         int extra_ref = 0;
4022 
4023         /*
4024          * This node is blocking another node from getting a read
4025          * lock. This happens when we've renamed within a
4026          * directory. We've forced the other nodes to d_delete(), but
4027          * we never actually dropped our lock because it's still
4028          * valid. The downconvert code will retain a PR for this node,
4029          * so there's no further work to do.
4030          */
4031         if (blocking == DLM_LOCK_PR)
4032                 return UNBLOCK_CONTINUE;
4033 
4034         /*
4035          * Mark this inode as potentially orphaned. The code in
4036          * ocfs2_delete_inode() will figure out whether it actually
4037          * needs to be freed or not.
4038          */
4039         spin_lock(&oi->ip_lock);
4040         oi->ip_flags |= OCFS2_INODE_MAYBE_ORPHANED;
4041         spin_unlock(&oi->ip_lock);
4042 
4043         /*
4044          * Yuck. We need to make sure however that the check of
4045          * OCFS2_LOCK_FREEING and the extra reference are atomic with
4046          * respect to a reference decrement or the setting of that
4047          * flag.
4048          */
4049         spin_lock_irqsave(&lockres->l_lock, flags);
4050         spin_lock(&dentry_attach_lock);
4051         if (!(lockres->l_flags & OCFS2_LOCK_FREEING)
4052             && dl->dl_count) {
4053                 dl->dl_count++;
4054                 extra_ref = 1;
4055         }
4056         spin_unlock(&dentry_attach_lock);
4057         spin_unlock_irqrestore(&lockres->l_lock, flags);
4058 
4059         mlog(0, "extra_ref = %d\n", extra_ref);
4060 
4061         /*
4062          * We have a process waiting on us in ocfs2_dentry_iput(),
4063          * which means we can't have any more outstanding
4064          * aliases. There's no need to do any more work.
4065          */
4066         if (!extra_ref)
4067                 return UNBLOCK_CONTINUE;
4068 
4069         spin_lock(&dentry_attach_lock);
4070         while (1) {
4071                 dentry = ocfs2_find_local_alias(dl->dl_inode,
4072                                                 dl->dl_parent_blkno, 1);
4073                 if (!dentry)
4074                         break;
4075                 spin_unlock(&dentry_attach_lock);
4076 
4077                 if (S_ISDIR(dl->dl_inode->i_mode))
4078                         shrink_dcache_parent(dentry);
4079 
4080                 mlog(0, "d_delete(%pd);\n", dentry);
4081 
4082                 /*
4083                  * The following dcache calls may do an
4084                  * iput(). Normally we don't want that from the
4085                  * downconverting thread, but in this case it's ok
4086                  * because the requesting node already has an
4087                  * exclusive lock on the inode, so it can't be queued
4088                  * for a downconvert.
4089                  */
4090                 d_delete(dentry);
4091                 dput(dentry);
4092 
4093                 spin_lock(&dentry_attach_lock);
4094         }
4095         spin_unlock(&dentry_attach_lock);
4096 
4097         /*
4098          * If we are the last holder of this dentry lock, there is no
4099          * reason to downconvert so skip straight to the unlock.
4100          */
4101         if (dl->dl_count == 1)
4102                 return UNBLOCK_STOP_POST;
4103 
4104         return UNBLOCK_CONTINUE_POST;
4105 }
4106 
4107 static int ocfs2_check_refcount_downconvert(struct ocfs2_lock_res *lockres,
4108                                             int new_level)
4109 {
4110         struct ocfs2_refcount_tree *tree =
4111                                 ocfs2_lock_res_refcount_tree(lockres);
4112 
4113         return ocfs2_ci_checkpointed(&tree->rf_ci, lockres, new_level);
4114 }
4115 
4116 static int ocfs2_refcount_convert_worker(struct ocfs2_lock_res *lockres,
4117                                          int blocking)
4118 {
4119         struct ocfs2_refcount_tree *tree =
4120                                 ocfs2_lock_res_refcount_tree(lockres);
4121 
4122         ocfs2_metadata_cache_purge(&tree->rf_ci);
4123 
4124         return UNBLOCK_CONTINUE;
4125 }
4126 
4127 static void ocfs2_set_qinfo_lvb(struct ocfs2_lock_res *lockres)
4128 {
4129         struct ocfs2_qinfo_lvb *lvb;
4130         struct ocfs2_mem_dqinfo *oinfo = ocfs2_lock_res_qinfo(lockres);
4131         struct mem_dqinfo *info = sb_dqinfo(oinfo->dqi_gi.dqi_sb,
4132                                             oinfo->dqi_gi.dqi_type);
4133 
4134         lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
4135         lvb->lvb_version = OCFS2_QINFO_LVB_VERSION;
4136         lvb->lvb_bgrace = cpu_to_be32(info->dqi_bgrace);
4137         lvb->lvb_igrace = cpu_to_be32(info->dqi_igrace);
4138         lvb->lvb_syncms = cpu_to_be32(oinfo->dqi_syncms);
4139         lvb->lvb_blocks = cpu_to_be32(oinfo->dqi_gi.dqi_blocks);
4140         lvb->lvb_free_blk = cpu_to_be32(oinfo->dqi_gi.dqi_free_blk);
4141         lvb->lvb_free_entry = cpu_to_be32(oinfo->dqi_gi.dqi_free_entry);
4142 }
4143 
4144 void ocfs2_qinfo_unlock(struct ocfs2_mem_dqinfo *oinfo, int ex)
4145 {
4146         struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
4147         struct ocfs2_super *osb = OCFS2_SB(oinfo->dqi_gi.dqi_sb);
4148         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4149 
4150         if (!ocfs2_is_hard_readonly(osb) && !ocfs2_mount_local(osb))
4151                 ocfs2_cluster_unlock(osb, lockres, level);
4152 }
4153 
4154 static int ocfs2_refresh_qinfo(struct ocfs2_mem_dqinfo *oinfo)
4155 {
4156         struct mem_dqinfo *info = sb_dqinfo(oinfo->dqi_gi.dqi_sb,
4157                                             oinfo->dqi_gi.dqi_type);
4158         struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
4159         struct ocfs2_qinfo_lvb *lvb = ocfs2_dlm_lvb(&lockres->l_lksb);
4160         struct buffer_head *bh = NULL;
4161         struct ocfs2_global_disk_dqinfo *gdinfo;
4162         int status = 0;
4163 
4164         if (ocfs2_dlm_lvb_valid(&lockres->l_lksb) &&
4165             lvb->lvb_version == OCFS2_QINFO_LVB_VERSION) {
4166                 info->dqi_bgrace = be32_to_cpu(lvb->lvb_bgrace);
4167                 info->dqi_igrace = be32_to_cpu(lvb->lvb_igrace);
4168                 oinfo->dqi_syncms = be32_to_cpu(lvb->lvb_syncms);
4169                 oinfo->dqi_gi.dqi_blocks = be32_to_cpu(lvb->lvb_blocks);
4170                 oinfo->dqi_gi.dqi_free_blk = be32_to_cpu(lvb->lvb_free_blk);
4171                 oinfo->dqi_gi.dqi_free_entry =
4172                                         be32_to_cpu(lvb->lvb_free_entry);
4173         } else {
4174                 status = ocfs2_read_quota_phys_block(oinfo->dqi_gqinode,
4175                                                      oinfo->dqi_giblk, &bh);
4176                 if (status) {
4177                         mlog_errno(status);
4178                         goto bail;
4179                 }
4180                 gdinfo = (struct ocfs2_global_disk_dqinfo *)
4181                                         (bh->b_data + OCFS2_GLOBAL_INFO_OFF);
4182                 info->dqi_bgrace = le32_to_cpu(gdinfo->dqi_bgrace);
4183                 info->dqi_igrace = le32_to_cpu(gdinfo->dqi_igrace);
4184                 oinfo->dqi_syncms = le32_to_cpu(gdinfo->dqi_syncms);
4185                 oinfo->dqi_gi.dqi_blocks = le32_to_cpu(gdinfo->dqi_blocks);
4186                 oinfo->dqi_gi.dqi_free_blk = le32_to_cpu(gdinfo->dqi_free_blk);
4187                 oinfo->dqi_gi.dqi_free_entry =
4188                                         le32_to_cpu(gdinfo->dqi_free_entry);
4189                 brelse(bh);
4190                 ocfs2_track_lock_refresh(lockres);
4191         }
4192 
4193 bail:
4194         return status;
4195 }
4196 
4197 /* Lock quota info, this function expects at least shared lock on the quota file
4198  * so that we can safely refresh quota info from disk. */
4199 int ocfs2_qinfo_lock(struct ocfs2_mem_dqinfo *oinfo, int ex)
4200 {
4201         struct ocfs2_lock_res *lockres = &oinfo->dqi_gqlock;
4202         struct ocfs2_super *osb = OCFS2_SB(oinfo->dqi_gi.dqi_sb);
4203         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4204         int status = 0;
4205 
4206         /* On RO devices, locking really isn't needed... */
4207         if (ocfs2_is_hard_readonly(osb)) {
4208                 if (ex)
4209                         status = -EROFS;
4210                 goto bail;
4211         }
4212         if (ocfs2_mount_local(osb))
4213                 goto bail;
4214 
4215         status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
4216         if (status < 0) {
4217                 mlog_errno(status);
4218                 goto bail;
4219         }
4220         if (!ocfs2_should_refresh_lock_res(lockres))
4221                 goto bail;
4222         /* OK, we have the lock but we need to refresh the quota info */
4223         status = ocfs2_refresh_qinfo(oinfo);
4224         if (status)
4225                 ocfs2_qinfo_unlock(oinfo, ex);
4226         ocfs2_complete_lock_res_refresh(lockres, status);
4227 bail:
4228         return status;
4229 }
4230 
4231 int ocfs2_refcount_lock(struct ocfs2_refcount_tree *ref_tree, int ex)
4232 {
4233         int status;
4234         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4235         struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
4236         struct ocfs2_super *osb = lockres->l_priv;
4237 
4238 
4239         if (ocfs2_is_hard_readonly(osb))
4240                 return -EROFS;
4241 
4242         if (ocfs2_mount_local(osb))
4243                 return 0;
4244 
4245         status = ocfs2_cluster_lock(osb, lockres, level, 0, 0);
4246         if (status < 0)
4247                 mlog_errno(status);
4248 
4249         return status;
4250 }
4251 
4252 void ocfs2_refcount_unlock(struct ocfs2_refcount_tree *ref_tree, int ex)
4253 {
4254         int level = ex ? DLM_LOCK_EX : DLM_LOCK_PR;
4255         struct ocfs2_lock_res *lockres = &ref_tree->rf_lockres;
4256         struct ocfs2_super *osb = lockres->l_priv;
4257 
4258         if (!ocfs2_mount_local(osb))
4259                 ocfs2_cluster_unlock(osb, lockres, level);
4260 }
4261 
4262 static void ocfs2_process_blocked_lock(struct ocfs2_super *osb,
4263                                        struct ocfs2_lock_res *lockres)
4264 {
4265         int status;
4266         struct ocfs2_unblock_ctl ctl = {0, 0,};
4267         unsigned long flags;
4268 
4269         /* Our reference to the lockres in this function can be
4270          * considered valid until we remove the OCFS2_LOCK_QUEUED
4271          * flag. */
4272 
4273         BUG_ON(!lockres);
4274         BUG_ON(!lockres->l_ops);
4275 
4276         mlog(ML_BASTS, "lockres %s blocked\n", lockres->l_name);
4277 
4278         /* Detect whether a lock has been marked as going away while
4279          * the downconvert thread was processing other things. A lock can
4280          * still be marked with OCFS2_LOCK_FREEING after this check,
4281          * but short circuiting here will still save us some
4282          * performance. */
4283         spin_lock_irqsave(&lockres->l_lock, flags);
4284         if (lockres->l_flags & OCFS2_LOCK_FREEING)
4285                 goto unqueue;
4286         spin_unlock_irqrestore(&lockres->l_lock, flags);
4287 
4288         status = ocfs2_unblock_lock(osb, lockres, &ctl);
4289         if (status < 0)
4290                 mlog_errno(status);
4291 
4292         spin_lock_irqsave(&lockres->l_lock, flags);
4293 unqueue:
4294         if (lockres->l_flags & OCFS2_LOCK_FREEING || !ctl.requeue) {
4295                 lockres_clear_flags(lockres, OCFS2_LOCK_QUEUED);
4296         } else
4297                 ocfs2_schedule_blocked_lock(osb, lockres);
4298 
4299         mlog(ML_BASTS, "lockres %s, requeue = %s.\n", lockres->l_name,
4300              ctl.requeue ? "yes" : "no");
4301         spin_unlock_irqrestore(&lockres->l_lock, flags);
4302 
4303         if (ctl.unblock_action != UNBLOCK_CONTINUE
4304             && lockres->l_ops->post_unlock)
4305                 lockres->l_ops->post_unlock(osb, lockres);
4306 }
4307 
4308 static void ocfs2_schedule_blocked_lock(struct ocfs2_super *osb,
4309                                         struct ocfs2_lock_res *lockres)
4310 {
4311         unsigned long flags;
4312 
4313         assert_spin_locked(&lockres->l_lock);
4314 
4315         if (lockres->l_flags & OCFS2_LOCK_FREEING) {
4316                 /* Do not schedule a lock for downconvert when it's on
4317                  * the way to destruction - any nodes wanting access
4318                  * to the resource will get it soon. */
4319                 mlog(ML_BASTS, "lockres %s won't be scheduled: flags 0x%lx\n",
4320                      lockres->l_name, lockres->l_flags);
4321                 return;
4322         }
4323 
4324         lockres_or_flags(lockres, OCFS2_LOCK_QUEUED);
4325 
4326         spin_lock_irqsave(&osb->dc_task_lock, flags);
4327         if (list_empty(&lockres->l_blocked_list)) {
4328                 list_add_tail(&lockres->l_blocked_list,
4329                               &osb->blocked_lock_list);
4330                 osb->blocked_lock_count++;
4331         }
4332         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4333 }
4334 
4335 static void ocfs2_downconvert_thread_do_work(struct ocfs2_super *osb)
4336 {
4337         unsigned long processed;
4338         unsigned long flags;
4339         struct ocfs2_lock_res *lockres;
4340 
4341         spin_lock_irqsave(&osb->dc_task_lock, flags);
4342         /* grab this early so we know to try again if a state change and
4343          * wake happens part-way through our work  */
4344         osb->dc_work_sequence = osb->dc_wake_sequence;
4345 
4346         processed = osb->blocked_lock_count;
4347         /*
4348          * blocked lock processing in this loop might call iput which can
4349          * remove items off osb->blocked_lock_list. Downconvert up to
4350          * 'processed' number of locks, but stop short if we had some
4351          * removed in ocfs2_mark_lockres_freeing when downconverting.
4352          */
4353         while (processed && !list_empty(&osb->blocked_lock_list)) {
4354                 lockres = list_entry(osb->blocked_lock_list.next,
4355                                      struct ocfs2_lock_res, l_blocked_list);
4356                 list_del_init(&lockres->l_blocked_list);
4357                 osb->blocked_lock_count--;
4358                 spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4359 
4360                 BUG_ON(!processed);
4361                 processed--;
4362 
4363                 ocfs2_process_blocked_lock(osb, lockres);
4364 
4365                 spin_lock_irqsave(&osb->dc_task_lock, flags);
4366         }
4367         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4368 }
4369 
4370 static int ocfs2_downconvert_thread_lists_empty(struct ocfs2_super *osb)
4371 {
4372         int empty = 0;
4373         unsigned long flags;
4374 
4375         spin_lock_irqsave(&osb->dc_task_lock, flags);
4376         if (list_empty(&osb->blocked_lock_list))
4377                 empty = 1;
4378 
4379         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4380         return empty;
4381 }
4382 
4383 static int ocfs2_downconvert_thread_should_wake(struct ocfs2_super *osb)
4384 {
4385         int should_wake = 0;
4386         unsigned long flags;
4387 
4388         spin_lock_irqsave(&osb->dc_task_lock, flags);
4389         if (osb->dc_work_sequence != osb->dc_wake_sequence)
4390                 should_wake = 1;
4391         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4392 
4393         return should_wake;
4394 }
4395 
4396 static int ocfs2_downconvert_thread(void *arg)
4397 {
4398         struct ocfs2_super *osb = arg;
4399 
4400         /* only quit once we've been asked to stop and there is no more
4401          * work available */
4402         while (!(kthread_should_stop() &&
4403                 ocfs2_downconvert_thread_lists_empty(osb))) {
4404 
4405                 wait_event_interruptible(osb->dc_event,
4406                                          ocfs2_downconvert_thread_should_wake(osb) ||
4407                                          kthread_should_stop());
4408 
4409                 mlog(0, "downconvert_thread: awoken\n");
4410 
4411                 ocfs2_downconvert_thread_do_work(osb);
4412         }
4413 
4414         osb->dc_task = NULL;
4415         return 0;
4416 }
4417 
4418 void ocfs2_wake_downconvert_thread(struct ocfs2_super *osb)
4419 {
4420         unsigned long flags;
4421 
4422         spin_lock_irqsave(&osb->dc_task_lock, flags);
4423         /* make sure the voting thread gets a swipe at whatever changes
4424          * the caller may have made to the voting state */
4425         osb->dc_wake_sequence++;
4426         spin_unlock_irqrestore(&osb->dc_task_lock, flags);
4427         wake_up(&osb->dc_event);
4428 }