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