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