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