fs/ocfs2/dlmglue.c

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