1// SPDX-License-Identifier: GPL-2.0-only
   2/******************************************************************************
   3*******************************************************************************
   4**
   5**  Copyright (C) 2005-2010 Red Hat, Inc.  All rights reserved.
   6**
 
 
 
   7**
   8*******************************************************************************
   9******************************************************************************/
  10
  11/* Central locking logic has four stages:
  12
  13   dlm_lock()
  14   dlm_unlock()
  15
  16   request_lock(ls, lkb)
  17   convert_lock(ls, lkb)
  18   unlock_lock(ls, lkb)
  19   cancel_lock(ls, lkb)
  20
  21   _request_lock(r, lkb)
  22   _convert_lock(r, lkb)
  23   _unlock_lock(r, lkb)
  24   _cancel_lock(r, lkb)
  25
  26   do_request(r, lkb)
  27   do_convert(r, lkb)
  28   do_unlock(r, lkb)
  29   do_cancel(r, lkb)
  30
  31   Stage 1 (lock, unlock) is mainly about checking input args and
  32   splitting into one of the four main operations:
  33
  34       dlm_lock          = request_lock
  35       dlm_lock+CONVERT  = convert_lock
  36       dlm_unlock        = unlock_lock
  37       dlm_unlock+CANCEL = cancel_lock
  38
  39   Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is
  40   provided to the next stage.
  41
  42   Stage 3, _xxxx_lock(), determines if the operation is local or remote.
  43   When remote, it calls send_xxxx(), when local it calls do_xxxx().
  44
  45   Stage 4, do_xxxx(), is the guts of the operation.  It manipulates the
  46   given rsb and lkb and queues callbacks.
  47
  48   For remote operations, send_xxxx() results in the corresponding do_xxxx()
  49   function being executed on the remote node.  The connecting send/receive
  50   calls on local (L) and remote (R) nodes:
  51
  52   L: send_xxxx()              ->  R: receive_xxxx()
  53                                   R: do_xxxx()
  54   L: receive_xxxx_reply()     <-  R: send_xxxx_reply()
  55*/
  56#include <trace/events/dlm.h>
  57
  58#include <linux/types.h>
  59#include <linux/rbtree.h>
  60#include <linux/slab.h>
  61#include "dlm_internal.h"
  62#include <linux/dlm_device.h>
  63#include "memory.h"
  64#include "midcomms.h"
  65#include "requestqueue.h"
  66#include "util.h"
  67#include "dir.h"
  68#include "member.h"
  69#include "lockspace.h"
  70#include "ast.h"
  71#include "lock.h"
  72#include "rcom.h"
  73#include "recover.h"
  74#include "lvb_table.h"
  75#include "user.h"
  76#include "config.h"
  77
  78static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb);
  79static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb);
  80static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  81static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb);
  82static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb);
  83static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode);
  84static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb);
  85static int send_remove(struct dlm_rsb *r);
  86static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  87static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb);
  88static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
  89				    struct dlm_message *ms);
  90static int receive_extralen(struct dlm_message *ms);
  91static void do_purge(struct dlm_ls *ls, int nodeid, int pid);
  92static void del_timeout(struct dlm_lkb *lkb);
  93static void toss_rsb(struct kref *kref);
  94
  95/*
  96 * Lock compatibilty matrix - thanks Steve
  97 * UN = Unlocked state. Not really a state, used as a flag
  98 * PD = Padding. Used to make the matrix a nice power of two in size
  99 * Other states are the same as the VMS DLM.
 100 * Usage: matrix[grmode+1][rqmode+1]  (although m[rq+1][gr+1] is the same)
 101 */
 102
 103static const int __dlm_compat_matrix[8][8] = {
 104      /* UN NL CR CW PR PW EX PD */
 105        {1, 1, 1, 1, 1, 1, 1, 0},       /* UN */
 106        {1, 1, 1, 1, 1, 1, 1, 0},       /* NL */
 107        {1, 1, 1, 1, 1, 1, 0, 0},       /* CR */
 108        {1, 1, 1, 1, 0, 0, 0, 0},       /* CW */
 109        {1, 1, 1, 0, 1, 0, 0, 0},       /* PR */
 110        {1, 1, 1, 0, 0, 0, 0, 0},       /* PW */
 111        {1, 1, 0, 0, 0, 0, 0, 0},       /* EX */
 112        {0, 0, 0, 0, 0, 0, 0, 0}        /* PD */
 113};
 114
 115/*
 116 * This defines the direction of transfer of LVB data.
 117 * Granted mode is the row; requested mode is the column.
 118 * Usage: matrix[grmode+1][rqmode+1]
 119 * 1 = LVB is returned to the caller
 120 * 0 = LVB is written to the resource
 121 * -1 = nothing happens to the LVB
 122 */
 123
 124const int dlm_lvb_operations[8][8] = {
 125        /* UN   NL  CR  CW  PR  PW  EX  PD*/
 126        {  -1,  1,  1,  1,  1,  1,  1, -1 }, /* UN */
 127        {  -1,  1,  1,  1,  1,  1,  1,  0 }, /* NL */
 128        {  -1, -1,  1,  1,  1,  1,  1,  0 }, /* CR */
 129        {  -1, -1, -1,  1,  1,  1,  1,  0 }, /* CW */
 130        {  -1, -1, -1, -1,  1,  1,  1,  0 }, /* PR */
 131        {  -1,  0,  0,  0,  0,  0,  1,  0 }, /* PW */
 132        {  -1,  0,  0,  0,  0,  0,  0,  0 }, /* EX */
 133        {  -1,  0,  0,  0,  0,  0,  0,  0 }  /* PD */
 134};
 135
 136#define modes_compat(gr, rq) \
 137	__dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1]
 138
 139int dlm_modes_compat(int mode1, int mode2)
 140{
 141	return __dlm_compat_matrix[mode1 + 1][mode2 + 1];
 142}
 143
 144/*
 145 * Compatibility matrix for conversions with QUECVT set.
 146 * Granted mode is the row; requested mode is the column.
 147 * Usage: matrix[grmode+1][rqmode+1]
 148 */
 149
 150static const int __quecvt_compat_matrix[8][8] = {
 151      /* UN NL CR CW PR PW EX PD */
 152        {0, 0, 0, 0, 0, 0, 0, 0},       /* UN */
 153        {0, 0, 1, 1, 1, 1, 1, 0},       /* NL */
 154        {0, 0, 0, 1, 1, 1, 1, 0},       /* CR */
 155        {0, 0, 0, 0, 1, 1, 1, 0},       /* CW */
 156        {0, 0, 0, 1, 0, 1, 1, 0},       /* PR */
 157        {0, 0, 0, 0, 0, 0, 1, 0},       /* PW */
 158        {0, 0, 0, 0, 0, 0, 0, 0},       /* EX */
 159        {0, 0, 0, 0, 0, 0, 0, 0}        /* PD */
 160};
 161
 162void dlm_print_lkb(struct dlm_lkb *lkb)
 163{
 164	printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x "
 165	       "sts %d rq %d gr %d wait_type %d wait_nodeid %d seq %llu\n",
 166	       lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags,
 167	       lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode,
 168	       lkb->lkb_grmode, lkb->lkb_wait_type, lkb->lkb_wait_nodeid,
 169	       (unsigned long long)lkb->lkb_recover_seq);
 170}
 171
 172static void dlm_print_rsb(struct dlm_rsb *r)
 173{
 174	printk(KERN_ERR "rsb: nodeid %d master %d dir %d flags %lx first %x "
 175	       "rlc %d name %s\n",
 176	       r->res_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
 177	       r->res_flags, r->res_first_lkid, r->res_recover_locks_count,
 178	       r->res_name);
 179}
 180
 181void dlm_dump_rsb(struct dlm_rsb *r)
 182{
 183	struct dlm_lkb *lkb;
 184
 185	dlm_print_rsb(r);
 186
 187	printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n",
 188	       list_empty(&r->res_root_list), list_empty(&r->res_recover_list));
 189	printk(KERN_ERR "rsb lookup list\n");
 190	list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup)
 191		dlm_print_lkb(lkb);
 192	printk(KERN_ERR "rsb grant queue:\n");
 193	list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue)
 194		dlm_print_lkb(lkb);
 195	printk(KERN_ERR "rsb convert queue:\n");
 196	list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue)
 197		dlm_print_lkb(lkb);
 198	printk(KERN_ERR "rsb wait queue:\n");
 199	list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue)
 200		dlm_print_lkb(lkb);
 201}
 202
 203/* Threads cannot use the lockspace while it's being recovered */
 204
 205static inline void dlm_lock_recovery(struct dlm_ls *ls)
 206{
 207	down_read(&ls->ls_in_recovery);
 208}
 209
 210void dlm_unlock_recovery(struct dlm_ls *ls)
 211{
 212	up_read(&ls->ls_in_recovery);
 213}
 214
 215int dlm_lock_recovery_try(struct dlm_ls *ls)
 216{
 217	return down_read_trylock(&ls->ls_in_recovery);
 218}
 219
 220static inline int can_be_queued(struct dlm_lkb *lkb)
 221{
 222	return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE);
 223}
 224
 225static inline int force_blocking_asts(struct dlm_lkb *lkb)
 226{
 227	return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST);
 228}
 229
 230static inline int is_demoted(struct dlm_lkb *lkb)
 231{
 232	return (lkb->lkb_sbflags & DLM_SBF_DEMOTED);
 233}
 234
 235static inline int is_altmode(struct dlm_lkb *lkb)
 236{
 237	return (lkb->lkb_sbflags & DLM_SBF_ALTMODE);
 238}
 239
 240static inline int is_granted(struct dlm_lkb *lkb)
 241{
 242	return (lkb->lkb_status == DLM_LKSTS_GRANTED);
 243}
 244
 245static inline int is_remote(struct dlm_rsb *r)
 246{
 247	DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r););
 248	return !!r->res_nodeid;
 249}
 250
 251static inline int is_process_copy(struct dlm_lkb *lkb)
 252{
 253	return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY));
 254}
 255
 256static inline int is_master_copy(struct dlm_lkb *lkb)
 257{
 258	return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0;
 259}
 260
 261static inline int middle_conversion(struct dlm_lkb *lkb)
 262{
 263	if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) ||
 264	    (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW))
 265		return 1;
 266	return 0;
 267}
 268
 269static inline int down_conversion(struct dlm_lkb *lkb)
 270{
 271	return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode);
 272}
 273
 274static inline int is_overlap_unlock(struct dlm_lkb *lkb)
 275{
 276	return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK;
 277}
 278
 279static inline int is_overlap_cancel(struct dlm_lkb *lkb)
 280{
 281	return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL;
 282}
 283
 284static inline int is_overlap(struct dlm_lkb *lkb)
 285{
 286	return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK |
 287				  DLM_IFL_OVERLAP_CANCEL));
 288}
 289
 290static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
 291{
 292	if (is_master_copy(lkb))
 293		return;
 294
 295	del_timeout(lkb);
 296
 297	DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb););
 298
 299#ifdef CONFIG_DLM_DEPRECATED_API
 300	/* if the operation was a cancel, then return -DLM_ECANCEL, if a
 301	   timeout caused the cancel then return -ETIMEDOUT */
 302	if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) {
 303		lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL;
 304		rv = -ETIMEDOUT;
 305	}
 306#endif
 307
 308	if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) {
 309		lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL;
 310		rv = -EDEADLK;
 311	}
 312
 313	dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, lkb->lkb_sbflags);
 314}
 315
 316static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb)
 317{
 318	queue_cast(r, lkb,
 319		   is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL);
 320}
 321
 322static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode)
 323{
 324	if (is_master_copy(lkb)) {
 325		send_bast(r, lkb, rqmode);
 326	} else {
 327		dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0);
 328	}
 329}
 330
 331/*
 332 * Basic operations on rsb's and lkb's
 333 */
 334
 335/* This is only called to add a reference when the code already holds
 336   a valid reference to the rsb, so there's no need for locking. */
 337
 338static inline void hold_rsb(struct dlm_rsb *r)
 339{
 340	kref_get(&r->res_ref);
 341}
 342
 343void dlm_hold_rsb(struct dlm_rsb *r)
 344{
 345	hold_rsb(r);
 346}
 347
 348/* When all references to the rsb are gone it's transferred to
 349   the tossed list for later disposal. */
 350
 351static void put_rsb(struct dlm_rsb *r)
 352{
 353	struct dlm_ls *ls = r->res_ls;
 354	uint32_t bucket = r->res_bucket;
 355	int rv;
 356
 357	rv = kref_put_lock(&r->res_ref, toss_rsb,
 358			   &ls->ls_rsbtbl[bucket].lock);
 359	if (rv)
 360		spin_unlock(&ls->ls_rsbtbl[bucket].lock);
 361}
 362
 363void dlm_put_rsb(struct dlm_rsb *r)
 364{
 365	put_rsb(r);
 366}
 367
 368static int pre_rsb_struct(struct dlm_ls *ls)
 369{
 370	struct dlm_rsb *r1, *r2;
 371	int count = 0;
 372
 373	spin_lock(&ls->ls_new_rsb_spin);
 374	if (ls->ls_new_rsb_count > dlm_config.ci_new_rsb_count / 2) {
 375		spin_unlock(&ls->ls_new_rsb_spin);
 376		return 0;
 377	}
 378	spin_unlock(&ls->ls_new_rsb_spin);
 379
 380	r1 = dlm_allocate_rsb(ls);
 381	r2 = dlm_allocate_rsb(ls);
 382
 383	spin_lock(&ls->ls_new_rsb_spin);
 384	if (r1) {
 385		list_add(&r1->res_hashchain, &ls->ls_new_rsb);
 386		ls->ls_new_rsb_count++;
 387	}
 388	if (r2) {
 389		list_add(&r2->res_hashchain, &ls->ls_new_rsb);
 390		ls->ls_new_rsb_count++;
 391	}
 392	count = ls->ls_new_rsb_count;
 393	spin_unlock(&ls->ls_new_rsb_spin);
 394
 395	if (!count)
 396		return -ENOMEM;
 397	return 0;
 398}
 399
 400/* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can
 401   unlock any spinlocks, go back and call pre_rsb_struct again.
 402   Otherwise, take an rsb off the list and return it. */
 403
 404static int get_rsb_struct(struct dlm_ls *ls, const void *name, int len,
 405			  struct dlm_rsb **r_ret)
 406{
 407	struct dlm_rsb *r;
 408	int count;
 409
 410	spin_lock(&ls->ls_new_rsb_spin);
 411	if (list_empty(&ls->ls_new_rsb)) {
 412		count = ls->ls_new_rsb_count;
 413		spin_unlock(&ls->ls_new_rsb_spin);
 414		log_debug(ls, "find_rsb retry %d %d %s",
 415			  count, dlm_config.ci_new_rsb_count,
 416			  (const char *)name);
 417		return -EAGAIN;
 418	}
 419
 420	r = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, res_hashchain);
 421	list_del(&r->res_hashchain);
 422	/* Convert the empty list_head to a NULL rb_node for tree usage: */
 423	memset(&r->res_hashnode, 0, sizeof(struct rb_node));
 424	ls->ls_new_rsb_count--;
 425	spin_unlock(&ls->ls_new_rsb_spin);
 426
 427	r->res_ls = ls;
 428	r->res_length = len;
 429	memcpy(r->res_name, name, len);
 430	mutex_init(&r->res_mutex);
 431
 432	INIT_LIST_HEAD(&r->res_lookup);
 433	INIT_LIST_HEAD(&r->res_grantqueue);
 434	INIT_LIST_HEAD(&r->res_convertqueue);
 435	INIT_LIST_HEAD(&r->res_waitqueue);
 436	INIT_LIST_HEAD(&r->res_root_list);
 437	INIT_LIST_HEAD(&r->res_recover_list);
 438
 439	*r_ret = r;
 440	return 0;
 441}
 442
 443static int rsb_cmp(struct dlm_rsb *r, const char *name, int nlen)
 444{
 445	char maxname[DLM_RESNAME_MAXLEN];
 446
 447	memset(maxname, 0, DLM_RESNAME_MAXLEN);
 448	memcpy(maxname, name, nlen);
 449	return memcmp(r->res_name, maxname, DLM_RESNAME_MAXLEN);
 450}
 451
 452int dlm_search_rsb_tree(struct rb_root *tree, const void *name, int len,
 453			struct dlm_rsb **r_ret)
 454{
 455	struct rb_node *node = tree->rb_node;
 456	struct dlm_rsb *r;
 
 457	int rc;
 458
 459	while (node) {
 460		r = rb_entry(node, struct dlm_rsb, res_hashnode);
 461		rc = rsb_cmp(r, name, len);
 462		if (rc < 0)
 463			node = node->rb_left;
 464		else if (rc > 0)
 465			node = node->rb_right;
 466		else
 467			goto found;
 468	}
 469	*r_ret = NULL;
 470	return -EBADR;
 471
 472 found:
 
 
 473	*r_ret = r;
 474	return 0;
 475}
 476
 477static int rsb_insert(struct dlm_rsb *rsb, struct rb_root *tree)
 478{
 479	struct rb_node **newn = &tree->rb_node;
 480	struct rb_node *parent = NULL;
 481	int rc;
 482
 483	while (*newn) {
 484		struct dlm_rsb *cur = rb_entry(*newn, struct dlm_rsb,
 485					       res_hashnode);
 486
 487		parent = *newn;
 488		rc = rsb_cmp(cur, rsb->res_name, rsb->res_length);
 489		if (rc < 0)
 490			newn = &parent->rb_left;
 491		else if (rc > 0)
 492			newn = &parent->rb_right;
 493		else {
 494			log_print("rsb_insert match");
 495			dlm_dump_rsb(rsb);
 496			dlm_dump_rsb(cur);
 497			return -EEXIST;
 498		}
 499	}
 500
 501	rb_link_node(&rsb->res_hashnode, parent, newn);
 502	rb_insert_color(&rsb->res_hashnode, tree);
 503	return 0;
 504}
 505
 506/*
 507 * Find rsb in rsbtbl and potentially create/add one
 508 *
 509 * Delaying the release of rsb's has a similar benefit to applications keeping
 510 * NL locks on an rsb, but without the guarantee that the cached master value
 511 * will still be valid when the rsb is reused.  Apps aren't always smart enough
 512 * to keep NL locks on an rsb that they may lock again shortly; this can lead
 513 * to excessive master lookups and removals if we don't delay the release.
 514 *
 515 * Searching for an rsb means looking through both the normal list and toss
 516 * list.  When found on the toss list the rsb is moved to the normal list with
 517 * ref count of 1; when found on normal list the ref count is incremented.
 518 *
 519 * rsb's on the keep list are being used locally and refcounted.
 520 * rsb's on the toss list are not being used locally, and are not refcounted.
 521 *
 522 * The toss list rsb's were either
 523 * - previously used locally but not any more (were on keep list, then
 524 *   moved to toss list when last refcount dropped)
 525 * - created and put on toss list as a directory record for a lookup
 526 *   (we are the dir node for the res, but are not using the res right now,
 527 *   but some other node is)
 528 *
 529 * The purpose of find_rsb() is to return a refcounted rsb for local use.
 530 * So, if the given rsb is on the toss list, it is moved to the keep list
 531 * before being returned.
 532 *
 533 * toss_rsb() happens when all local usage of the rsb is done, i.e. no
 534 * more refcounts exist, so the rsb is moved from the keep list to the
 535 * toss list.
 536 *
 537 * rsb's on both keep and toss lists are used for doing a name to master
 538 * lookups.  rsb's that are in use locally (and being refcounted) are on
 539 * the keep list, rsb's that are not in use locally (not refcounted) and
 540 * only exist for name/master lookups are on the toss list.
 541 *
 542 * rsb's on the toss list who's dir_nodeid is not local can have stale
 543 * name/master mappings.  So, remote requests on such rsb's can potentially
 544 * return with an error, which means the mapping is stale and needs to
 545 * be updated with a new lookup.  (The idea behind MASTER UNCERTAIN and
 546 * first_lkid is to keep only a single outstanding request on an rsb
 547 * while that rsb has a potentially stale master.)
 548 */
 549
 550static int find_rsb_dir(struct dlm_ls *ls, const void *name, int len,
 551			uint32_t hash, uint32_t b,
 552			int dir_nodeid, int from_nodeid,
 553			unsigned int flags, struct dlm_rsb **r_ret)
 554{
 555	struct dlm_rsb *r = NULL;
 556	int our_nodeid = dlm_our_nodeid();
 557	int from_local = 0;
 558	int from_other = 0;
 559	int from_dir = 0;
 560	int create = 0;
 561	int error;
 562
 563	if (flags & R_RECEIVE_REQUEST) {
 564		if (from_nodeid == dir_nodeid)
 565			from_dir = 1;
 566		else
 567			from_other = 1;
 568	} else if (flags & R_REQUEST) {
 569		from_local = 1;
 570	}
 571
 572	/*
 573	 * flags & R_RECEIVE_RECOVER is from dlm_recover_master_copy, so
 574	 * from_nodeid has sent us a lock in dlm_recover_locks, believing
 575	 * we're the new master.  Our local recovery may not have set
 576	 * res_master_nodeid to our_nodeid yet, so allow either.  Don't
 577	 * create the rsb; dlm_recover_process_copy() will handle EBADR
 578	 * by resending.
 579	 *
 580	 * If someone sends us a request, we are the dir node, and we do
 581	 * not find the rsb anywhere, then recreate it.  This happens if
 582	 * someone sends us a request after we have removed/freed an rsb
 583	 * from our toss list.  (They sent a request instead of lookup
 584	 * because they are using an rsb from their toss list.)
 585	 */
 586
 587	if (from_local || from_dir ||
 588	    (from_other && (dir_nodeid == our_nodeid))) {
 589		create = 1;
 590	}
 591
 592 retry:
 593	if (create) {
 594		error = pre_rsb_struct(ls);
 595		if (error < 0)
 596			goto out;
 597	}
 
 
 598
 599	spin_lock(&ls->ls_rsbtbl[b].lock);
 600
 601	error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
 602	if (error)
 603		goto do_toss;
 604	
 605	/*
 606	 * rsb is active, so we can't check master_nodeid without lock_rsb.
 607	 */
 608
 609	kref_get(&r->res_ref);
 610	goto out_unlock;
 611
 612
 613 do_toss:
 614	error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
 615	if (error)
 616		goto do_new;
 617
 618	/*
 619	 * rsb found inactive (master_nodeid may be out of date unless
 620	 * we are the dir_nodeid or were the master)  No other thread
 621	 * is using this rsb because it's on the toss list, so we can
 622	 * look at or update res_master_nodeid without lock_rsb.
 623	 */
 624
 625	if ((r->res_master_nodeid != our_nodeid) && from_other) {
 626		/* our rsb was not master, and another node (not the dir node)
 627		   has sent us a request */
 628		log_debug(ls, "find_rsb toss from_other %d master %d dir %d %s",
 629			  from_nodeid, r->res_master_nodeid, dir_nodeid,
 630			  r->res_name);
 631		error = -ENOTBLK;
 632		goto out_unlock;
 633	}
 634
 635	if ((r->res_master_nodeid != our_nodeid) && from_dir) {
 636		/* don't think this should ever happen */
 637		log_error(ls, "find_rsb toss from_dir %d master %d",
 638			  from_nodeid, r->res_master_nodeid);
 639		dlm_print_rsb(r);
 640		/* fix it and go on */
 641		r->res_master_nodeid = our_nodeid;
 642		r->res_nodeid = 0;
 643		rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
 644		r->res_first_lkid = 0;
 645	}
 646
 647	if (from_local && (r->res_master_nodeid != our_nodeid)) {
 648		/* Because we have held no locks on this rsb,
 649		   res_master_nodeid could have become stale. */
 650		rsb_set_flag(r, RSB_MASTER_UNCERTAIN);
 651		r->res_first_lkid = 0;
 652	}
 653
 654	rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
 655	error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 656	goto out_unlock;
 657
 658
 659 do_new:
 660	/*
 661	 * rsb not found
 662	 */
 663
 664	if (error == -EBADR && !create)
 665		goto out_unlock;
 666
 667	error = get_rsb_struct(ls, name, len, &r);
 668	if (error == -EAGAIN) {
 669		spin_unlock(&ls->ls_rsbtbl[b].lock);
 670		goto retry;
 671	}
 672	if (error)
 673		goto out_unlock;
 674
 675	r->res_hash = hash;
 676	r->res_bucket = b;
 677	r->res_dir_nodeid = dir_nodeid;
 678	kref_init(&r->res_ref);
 679
 680	if (from_dir) {
 681		/* want to see how often this happens */
 682		log_debug(ls, "find_rsb new from_dir %d recreate %s",
 683			  from_nodeid, r->res_name);
 684		r->res_master_nodeid = our_nodeid;
 685		r->res_nodeid = 0;
 686		goto out_add;
 687	}
 688
 689	if (from_other && (dir_nodeid != our_nodeid)) {
 690		/* should never happen */
 691		log_error(ls, "find_rsb new from_other %d dir %d our %d %s",
 692			  from_nodeid, dir_nodeid, our_nodeid, r->res_name);
 693		dlm_free_rsb(r);
 694		r = NULL;
 695		error = -ENOTBLK;
 696		goto out_unlock;
 697	}
 698
 699	if (from_other) {
 700		log_debug(ls, "find_rsb new from_other %d dir %d %s",
 701			  from_nodeid, dir_nodeid, r->res_name);
 702	}
 703
 704	if (dir_nodeid == our_nodeid) {
 705		/* When we are the dir nodeid, we can set the master
 706		   node immediately */
 707		r->res_master_nodeid = our_nodeid;
 708		r->res_nodeid = 0;
 709	} else {
 710		/* set_master will send_lookup to dir_nodeid */
 711		r->res_master_nodeid = 0;
 712		r->res_nodeid = -1;
 713	}
 714
 715 out_add:
 716	error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 717 out_unlock:
 718	spin_unlock(&ls->ls_rsbtbl[b].lock);
 719 out:
 720	*r_ret = r;
 721	return error;
 722}
 723
 724/* During recovery, other nodes can send us new MSTCPY locks (from
 725   dlm_recover_locks) before we've made ourself master (in
 726   dlm_recover_masters). */
 727
 728static int find_rsb_nodir(struct dlm_ls *ls, const void *name, int len,
 729			  uint32_t hash, uint32_t b,
 730			  int dir_nodeid, int from_nodeid,
 731			  unsigned int flags, struct dlm_rsb **r_ret)
 732{
 733	struct dlm_rsb *r = NULL;
 734	int our_nodeid = dlm_our_nodeid();
 735	int recover = (flags & R_RECEIVE_RECOVER);
 736	int error;
 737
 738 retry:
 739	error = pre_rsb_struct(ls);
 740	if (error < 0)
 741		goto out;
 742
 743	spin_lock(&ls->ls_rsbtbl[b].lock);
 744
 745	error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
 746	if (error)
 747		goto do_toss;
 748
 749	/*
 750	 * rsb is active, so we can't check master_nodeid without lock_rsb.
 751	 */
 752
 753	kref_get(&r->res_ref);
 754	goto out_unlock;
 755
 756
 757 do_toss:
 758	error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
 759	if (error)
 760		goto do_new;
 761
 762	/*
 763	 * rsb found inactive. No other thread is using this rsb because
 764	 * it's on the toss list, so we can look at or update
 765	 * res_master_nodeid without lock_rsb.
 766	 */
 767
 768	if (!recover && (r->res_master_nodeid != our_nodeid) && from_nodeid) {
 769		/* our rsb is not master, and another node has sent us a
 770		   request; this should never happen */
 771		log_error(ls, "find_rsb toss from_nodeid %d master %d dir %d",
 772			  from_nodeid, r->res_master_nodeid, dir_nodeid);
 773		dlm_print_rsb(r);
 774		error = -ENOTBLK;
 775		goto out_unlock;
 776	}
 777
 778	if (!recover && (r->res_master_nodeid != our_nodeid) &&
 779	    (dir_nodeid == our_nodeid)) {
 780		/* our rsb is not master, and we are dir; may as well fix it;
 781		   this should never happen */
 782		log_error(ls, "find_rsb toss our %d master %d dir %d",
 783			  our_nodeid, r->res_master_nodeid, dir_nodeid);
 784		dlm_print_rsb(r);
 785		r->res_master_nodeid = our_nodeid;
 786		r->res_nodeid = 0;
 787	}
 788
 789	rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
 790	error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 791	goto out_unlock;
 792
 793
 794 do_new:
 795	/*
 796	 * rsb not found
 797	 */
 798
 799	error = get_rsb_struct(ls, name, len, &r);
 800	if (error == -EAGAIN) {
 801		spin_unlock(&ls->ls_rsbtbl[b].lock);
 802		goto retry;
 803	}
 804	if (error)
 805		goto out_unlock;
 806
 807	r->res_hash = hash;
 808	r->res_bucket = b;
 809	r->res_dir_nodeid = dir_nodeid;
 810	r->res_master_nodeid = dir_nodeid;
 811	r->res_nodeid = (dir_nodeid == our_nodeid) ? 0 : dir_nodeid;
 812	kref_init(&r->res_ref);
 813
 814	error = rsb_insert(r, &ls->ls_rsbtbl[b].keep);
 815 out_unlock:
 816	spin_unlock(&ls->ls_rsbtbl[b].lock);
 817 out:
 818	*r_ret = r;
 819	return error;
 820}
 821
 822static int find_rsb(struct dlm_ls *ls, const void *name, int len,
 823		    int from_nodeid, unsigned int flags,
 824		    struct dlm_rsb **r_ret)
 825{
 826	uint32_t hash, b;
 827	int dir_nodeid;
 828
 829	if (len > DLM_RESNAME_MAXLEN)
 830		return -EINVAL;
 831
 832	hash = jhash(name, len, 0);
 833	b = hash & (ls->ls_rsbtbl_size - 1);
 834
 835	dir_nodeid = dlm_hash2nodeid(ls, hash);
 836
 837	if (dlm_no_directory(ls))
 838		return find_rsb_nodir(ls, name, len, hash, b, dir_nodeid,
 839				      from_nodeid, flags, r_ret);
 840	else
 841		return find_rsb_dir(ls, name, len, hash, b, dir_nodeid,
 842				      from_nodeid, flags, r_ret);
 843}
 844
 845/* we have received a request and found that res_master_nodeid != our_nodeid,
 846   so we need to return an error or make ourself the master */
 847
 848static int validate_master_nodeid(struct dlm_ls *ls, struct dlm_rsb *r,
 849				  int from_nodeid)
 850{
 851	if (dlm_no_directory(ls)) {
 852		log_error(ls, "find_rsb keep from_nodeid %d master %d dir %d",
 853			  from_nodeid, r->res_master_nodeid,
 854			  r->res_dir_nodeid);
 855		dlm_print_rsb(r);
 856		return -ENOTBLK;
 857	}
 858
 859	if (from_nodeid != r->res_dir_nodeid) {
 860		/* our rsb is not master, and another node (not the dir node)
 861	   	   has sent us a request.  this is much more common when our
 862	   	   master_nodeid is zero, so limit debug to non-zero.  */
 863
 864		if (r->res_master_nodeid) {
 865			log_debug(ls, "validate master from_other %d master %d "
 866				  "dir %d first %x %s", from_nodeid,
 867				  r->res_master_nodeid, r->res_dir_nodeid,
 868				  r->res_first_lkid, r->res_name);
 869		}
 870		return -ENOTBLK;
 871	} else {
 872		/* our rsb is not master, but the dir nodeid has sent us a
 873	   	   request; this could happen with master 0 / res_nodeid -1 */
 874
 875		if (r->res_master_nodeid) {
 876			log_error(ls, "validate master from_dir %d master %d "
 877				  "first %x %s",
 878				  from_nodeid, r->res_master_nodeid,
 879				  r->res_first_lkid, r->res_name);
 880		}
 881
 882		r->res_master_nodeid = dlm_our_nodeid();
 883		r->res_nodeid = 0;
 884		return 0;
 885	}
 886}
 887
 888static void __dlm_master_lookup(struct dlm_ls *ls, struct dlm_rsb *r, int our_nodeid,
 889				int from_nodeid, bool toss_list, unsigned int flags,
 890				int *r_nodeid, int *result)
 891{
 892	int fix_master = (flags & DLM_LU_RECOVER_MASTER);
 893	int from_master = (flags & DLM_LU_RECOVER_DIR);
 894
 895	if (r->res_dir_nodeid != our_nodeid) {
 896		/* should not happen, but may as well fix it and carry on */
 897		log_error(ls, "%s res_dir %d our %d %s", __func__,
 898			  r->res_dir_nodeid, our_nodeid, r->res_name);
 899		r->res_dir_nodeid = our_nodeid;
 900	}
 901
 902	if (fix_master && dlm_is_removed(ls, r->res_master_nodeid)) {
 903		/* Recovery uses this function to set a new master when
 904		 * the previous master failed.  Setting NEW_MASTER will
 905		 * force dlm_recover_masters to call recover_master on this
 906		 * rsb even though the res_nodeid is no longer removed.
 907		 */
 908
 909		r->res_master_nodeid = from_nodeid;
 910		r->res_nodeid = from_nodeid;
 911		rsb_set_flag(r, RSB_NEW_MASTER);
 912
 913		if (toss_list) {
 914			/* I don't think we should ever find it on toss list. */
 915			log_error(ls, "%s fix_master on toss", __func__);
 916			dlm_dump_rsb(r);
 917		}
 918	}
 919
 920	if (from_master && (r->res_master_nodeid != from_nodeid)) {
 921		/* this will happen if from_nodeid became master during
 922		 * a previous recovery cycle, and we aborted the previous
 923		 * cycle before recovering this master value
 924		 */
 925
 926		log_limit(ls, "%s from_master %d master_nodeid %d res_nodeid %d first %x %s",
 927			  __func__, from_nodeid, r->res_master_nodeid,
 928			  r->res_nodeid, r->res_first_lkid, r->res_name);
 929
 930		if (r->res_master_nodeid == our_nodeid) {
 931			log_error(ls, "from_master %d our_master", from_nodeid);
 932			dlm_dump_rsb(r);
 933			goto ret_assign;
 934		}
 935
 936		r->res_master_nodeid = from_nodeid;
 937		r->res_nodeid = from_nodeid;
 938		rsb_set_flag(r, RSB_NEW_MASTER);
 939	}
 940
 941	if (!r->res_master_nodeid) {
 942		/* this will happen if recovery happens while we're looking
 943		 * up the master for this rsb
 944		 */
 945
 946		log_debug(ls, "%s master 0 to %d first %x %s", __func__,
 947			  from_nodeid, r->res_first_lkid, r->res_name);
 948		r->res_master_nodeid = from_nodeid;
 949		r->res_nodeid = from_nodeid;
 950	}
 951
 952	if (!from_master && !fix_master &&
 953	    (r->res_master_nodeid == from_nodeid)) {
 954		/* this can happen when the master sends remove, the dir node
 955		 * finds the rsb on the keep list and ignores the remove,
 956		 * and the former master sends a lookup
 957		 */
 958
 959		log_limit(ls, "%s from master %d flags %x first %x %s",
 960			  __func__, from_nodeid, flags, r->res_first_lkid,
 961			  r->res_name);
 962	}
 963
 964 ret_assign:
 965	*r_nodeid = r->res_master_nodeid;
 966	if (result)
 967		*result = DLM_LU_MATCH;
 968}
 969
 970/*
 971 * We're the dir node for this res and another node wants to know the
 972 * master nodeid.  During normal operation (non recovery) this is only
 973 * called from receive_lookup(); master lookups when the local node is
 974 * the dir node are done by find_rsb().
 975 *
 976 * normal operation, we are the dir node for a resource
 977 * . _request_lock
 978 * . set_master
 979 * . send_lookup
 980 * . receive_lookup
 981 * . dlm_master_lookup flags 0
 982 *
 983 * recover directory, we are rebuilding dir for all resources
 984 * . dlm_recover_directory
 985 * . dlm_rcom_names
 986 *   remote node sends back the rsb names it is master of and we are dir of
 987 * . dlm_master_lookup RECOVER_DIR (fix_master 0, from_master 1)
 988 *   we either create new rsb setting remote node as master, or find existing
 989 *   rsb and set master to be the remote node.
 990 *
 991 * recover masters, we are finding the new master for resources
 992 * . dlm_recover_masters
 993 * . recover_master
 994 * . dlm_send_rcom_lookup
 995 * . receive_rcom_lookup
 996 * . dlm_master_lookup RECOVER_MASTER (fix_master 1, from_master 0)
 997 */
 998
 999int dlm_master_lookup(struct dlm_ls *ls, int from_nodeid, char *name, int len,
1000		      unsigned int flags, int *r_nodeid, int *result)
1001{
1002	struct dlm_rsb *r = NULL;
1003	uint32_t hash, b;
1004	int our_nodeid = dlm_our_nodeid();
1005	int dir_nodeid, error;
1006
1007	if (len > DLM_RESNAME_MAXLEN)
1008		return -EINVAL;
1009
1010	if (from_nodeid == our_nodeid) {
1011		log_error(ls, "dlm_master_lookup from our_nodeid %d flags %x",
1012			  our_nodeid, flags);
1013		return -EINVAL;
1014	}
1015
1016	hash = jhash(name, len, 0);
1017	b = hash & (ls->ls_rsbtbl_size - 1);
1018
1019	dir_nodeid = dlm_hash2nodeid(ls, hash);
1020	if (dir_nodeid != our_nodeid) {
1021		log_error(ls, "dlm_master_lookup from %d dir %d our %d h %x %d",
1022			  from_nodeid, dir_nodeid, our_nodeid, hash,
1023			  ls->ls_num_nodes);
1024		*r_nodeid = -1;
1025		return -EINVAL;
1026	}
1027
1028 retry:
1029	error = pre_rsb_struct(ls);
1030	if (error < 0)
1031		return error;
1032
1033	spin_lock(&ls->ls_rsbtbl[b].lock);
1034	error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
1035	if (!error) {
1036		/* because the rsb is active, we need to lock_rsb before
1037		 * checking/changing re_master_nodeid
1038		 */
1039
1040		hold_rsb(r);
1041		spin_unlock(&ls->ls_rsbtbl[b].lock);
1042		lock_rsb(r);
1043
1044		__dlm_master_lookup(ls, r, our_nodeid, from_nodeid, false,
1045				    flags, r_nodeid, result);
1046
1047		/* the rsb was active */
1048		unlock_rsb(r);
1049		put_rsb(r);
1050
1051		return 0;
1052	}
1053
1054	error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1055	if (error)
1056		goto not_found;
1057
1058	/* because the rsb is inactive (on toss list), it's not refcounted
1059	 * and lock_rsb is not used, but is protected by the rsbtbl lock
1060	 */
1061
1062	__dlm_master_lookup(ls, r, our_nodeid, from_nodeid, true, flags,
1063			    r_nodeid, result);
 
1064
1065	r->res_toss_time = jiffies;
1066	/* the rsb was inactive (on toss list) */
1067	spin_unlock(&ls->ls_rsbtbl[b].lock);
1068
1069	return 0;
 
 
1070
1071 not_found:
1072	error = get_rsb_struct(ls, name, len, &r);
1073	if (error == -EAGAIN) {
1074		spin_unlock(&ls->ls_rsbtbl[b].lock);
1075		goto retry;
1076	}
1077	if (error)
1078		goto out_unlock;
1079
1080	r->res_hash = hash;
1081	r->res_bucket = b;
1082	r->res_dir_nodeid = our_nodeid;
1083	r->res_master_nodeid = from_nodeid;
1084	r->res_nodeid = from_nodeid;
1085	kref_init(&r->res_ref);
1086	r->res_toss_time = jiffies;
1087
1088	error = rsb_insert(r, &ls->ls_rsbtbl[b].toss);
1089	if (error) {
1090		/* should never happen */
1091		dlm_free_rsb(r);
1092		spin_unlock(&ls->ls_rsbtbl[b].lock);
1093		goto retry;
1094	}
1095
1096	if (result)
1097		*result = DLM_LU_ADD;
1098	*r_nodeid = from_nodeid;
1099 out_unlock:
1100	spin_unlock(&ls->ls_rsbtbl[b].lock);
 
 
1101	return error;
1102}
1103
1104static void dlm_dump_rsb_hash(struct dlm_ls *ls, uint32_t hash)
1105{
1106	struct rb_node *n;
1107	struct dlm_rsb *r;
1108	int i;
1109
1110	for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1111		spin_lock(&ls->ls_rsbtbl[i].lock);
1112		for (n = rb_first(&ls->ls_rsbtbl[i].keep); n; n = rb_next(n)) {
1113			r = rb_entry(n, struct dlm_rsb, res_hashnode);
1114			if (r->res_hash == hash)
1115				dlm_dump_rsb(r);
1116		}
1117		spin_unlock(&ls->ls_rsbtbl[i].lock);
1118	}
1119}
1120
1121void dlm_dump_rsb_name(struct dlm_ls *ls, char *name, int len)
1122{
1123	struct dlm_rsb *r = NULL;
1124	uint32_t hash, b;
1125	int error;
1126
1127	hash = jhash(name, len, 0);
1128	b = hash & (ls->ls_rsbtbl_size - 1);
1129
1130	spin_lock(&ls->ls_rsbtbl[b].lock);
1131	error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
1132	if (!error)
1133		goto out_dump;
1134
1135	error = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1136	if (error)
1137		goto out;
1138 out_dump:
1139	dlm_dump_rsb(r);
1140 out:
1141	spin_unlock(&ls->ls_rsbtbl[b].lock);
1142}
1143
1144static void toss_rsb(struct kref *kref)
1145{
1146	struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1147	struct dlm_ls *ls = r->res_ls;
1148
1149	DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r););
1150	kref_init(&r->res_ref);
1151	rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[r->res_bucket].keep);
1152	rsb_insert(r, &ls->ls_rsbtbl[r->res_bucket].toss);
1153	r->res_toss_time = jiffies;
1154	ls->ls_rsbtbl[r->res_bucket].flags |= DLM_RTF_SHRINK;
1155	if (r->res_lvbptr) {
1156		dlm_free_lvb(r->res_lvbptr);
1157		r->res_lvbptr = NULL;
1158	}
1159}
1160
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1161/* See comment for unhold_lkb */
1162
1163static void unhold_rsb(struct dlm_rsb *r)
1164{
1165	int rv;
1166	rv = kref_put(&r->res_ref, toss_rsb);
1167	DLM_ASSERT(!rv, dlm_dump_rsb(r););
1168}
1169
1170static void kill_rsb(struct kref *kref)
1171{
1172	struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref);
1173
1174	/* All work is done after the return from kref_put() so we
1175	   can release the write_lock before the remove and free. */
1176
1177	DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r););
1178	DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r););
1179	DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r););
1180	DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r););
1181	DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r););
1182	DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r););
1183}
1184
1185/* Attaching/detaching lkb's from rsb's is for rsb reference counting.
1186   The rsb must exist as long as any lkb's for it do. */
1187
1188static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1189{
1190	hold_rsb(r);
1191	lkb->lkb_resource = r;
1192}
1193
1194static void detach_lkb(struct dlm_lkb *lkb)
1195{
1196	if (lkb->lkb_resource) {
1197		put_rsb(lkb->lkb_resource);
1198		lkb->lkb_resource = NULL;
1199	}
1200}
1201
1202static int _create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret,
1203		       int start, int end)
1204{
1205	struct dlm_lkb *lkb;
1206	int rv;
1207
1208	lkb = dlm_allocate_lkb(ls);
1209	if (!lkb)
1210		return -ENOMEM;
1211
1212	lkb->lkb_last_bast_mode = -1;
1213	lkb->lkb_nodeid = -1;
1214	lkb->lkb_grmode = DLM_LOCK_IV;
1215	kref_init(&lkb->lkb_ref);
1216	INIT_LIST_HEAD(&lkb->lkb_ownqueue);
1217	INIT_LIST_HEAD(&lkb->lkb_rsb_lookup);
1218#ifdef CONFIG_DLM_DEPRECATED_API
1219	INIT_LIST_HEAD(&lkb->lkb_time_list);
1220#endif
1221	INIT_LIST_HEAD(&lkb->lkb_cb_list);
1222	INIT_LIST_HEAD(&lkb->lkb_callbacks);
1223	spin_lock_init(&lkb->lkb_cb_lock);
1224	INIT_WORK(&lkb->lkb_cb_work, dlm_callback_work);
1225
1226	idr_preload(GFP_NOFS);
 
 
 
 
1227	spin_lock(&ls->ls_lkbidr_spin);
1228	rv = idr_alloc(&ls->ls_lkbidr, lkb, start, end, GFP_NOWAIT);
1229	if (rv >= 0)
1230		lkb->lkb_id = rv;
1231	spin_unlock(&ls->ls_lkbidr_spin);
1232	idr_preload_end();
 
 
1233
1234	if (rv < 0) {
1235		log_error(ls, "create_lkb idr error %d", rv);
1236		dlm_free_lkb(lkb);
1237		return rv;
1238	}
1239
1240	*lkb_ret = lkb;
1241	return 0;
1242}
1243
1244static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret)
1245{
1246	return _create_lkb(ls, lkb_ret, 1, 0);
1247}
1248
1249static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret)
1250{
1251	struct dlm_lkb *lkb;
1252
1253	spin_lock(&ls->ls_lkbidr_spin);
1254	lkb = idr_find(&ls->ls_lkbidr, lkid);
1255	if (lkb)
1256		kref_get(&lkb->lkb_ref);
1257	spin_unlock(&ls->ls_lkbidr_spin);
1258
1259	*lkb_ret = lkb;
1260	return lkb ? 0 : -ENOENT;
1261}
1262
1263static void kill_lkb(struct kref *kref)
1264{
1265	struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1266
1267	/* All work is done after the return from kref_put() so we
1268	   can release the write_lock before the detach_lkb */
1269
1270	DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1271}
1272
1273/* __put_lkb() is used when an lkb may not have an rsb attached to
1274   it so we need to provide the lockspace explicitly */
1275
1276static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb)
1277{
1278	uint32_t lkid = lkb->lkb_id;
1279	int rv;
1280
1281	rv = kref_put_lock(&lkb->lkb_ref, kill_lkb,
1282			   &ls->ls_lkbidr_spin);
1283	if (rv) {
1284		idr_remove(&ls->ls_lkbidr, lkid);
1285		spin_unlock(&ls->ls_lkbidr_spin);
1286
1287		detach_lkb(lkb);
1288
1289		/* for local/process lkbs, lvbptr points to caller's lksb */
1290		if (lkb->lkb_lvbptr && is_master_copy(lkb))
1291			dlm_free_lvb(lkb->lkb_lvbptr);
1292		dlm_free_lkb(lkb);
 
 
 
 
1293	}
1294
1295	return rv;
1296}
1297
1298int dlm_put_lkb(struct dlm_lkb *lkb)
1299{
1300	struct dlm_ls *ls;
1301
1302	DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb););
1303	DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb););
1304
1305	ls = lkb->lkb_resource->res_ls;
1306	return __put_lkb(ls, lkb);
1307}
1308
1309/* This is only called to add a reference when the code already holds
1310   a valid reference to the lkb, so there's no need for locking. */
1311
1312static inline void hold_lkb(struct dlm_lkb *lkb)
1313{
1314	kref_get(&lkb->lkb_ref);
1315}
1316
1317static void unhold_lkb_assert(struct kref *kref)
1318{
1319	struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref);
1320
1321	DLM_ASSERT(false, dlm_print_lkb(lkb););
1322}
1323
1324/* This is called when we need to remove a reference and are certain
1325   it's not the last ref.  e.g. del_lkb is always called between a
1326   find_lkb/put_lkb and is always the inverse of a previous add_lkb.
1327   put_lkb would work fine, but would involve unnecessary locking */
1328
1329static inline void unhold_lkb(struct dlm_lkb *lkb)
1330{
1331	kref_put(&lkb->lkb_ref, unhold_lkb_assert);
 
 
1332}
1333
1334static void lkb_add_ordered(struct list_head *new, struct list_head *head,
1335			    int mode)
1336{
1337	struct dlm_lkb *lkb = NULL, *iter;
1338
1339	list_for_each_entry(iter, head, lkb_statequeue)
1340		if (iter->lkb_rqmode < mode) {
1341			lkb = iter;
1342			list_add_tail(new, &iter->lkb_statequeue);
1343			break;
1344		}
1345
1346	if (!lkb)
1347		list_add_tail(new, head);
1348}
1349
1350/* add/remove lkb to rsb's grant/convert/wait queue */
1351
1352static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status)
1353{
1354	kref_get(&lkb->lkb_ref);
1355
1356	DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb););
1357
1358	lkb->lkb_timestamp = ktime_get();
1359
1360	lkb->lkb_status = status;
1361
1362	switch (status) {
1363	case DLM_LKSTS_WAITING:
1364		if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1365			list_add(&lkb->lkb_statequeue, &r->res_waitqueue);
1366		else
1367			list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue);
1368		break;
1369	case DLM_LKSTS_GRANTED:
1370		/* convention says granted locks kept in order of grmode */
1371		lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue,
1372				lkb->lkb_grmode);
1373		break;
1374	case DLM_LKSTS_CONVERT:
1375		if (lkb->lkb_exflags & DLM_LKF_HEADQUE)
1376			list_add(&lkb->lkb_statequeue, &r->res_convertqueue);
1377		else
1378			list_add_tail(&lkb->lkb_statequeue,
1379				      &r->res_convertqueue);
1380		break;
1381	default:
1382		DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status););
1383	}
1384}
1385
1386static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb)
1387{
1388	lkb->lkb_status = 0;
1389	list_del(&lkb->lkb_statequeue);
1390	unhold_lkb(lkb);
1391}
1392
1393static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts)
1394{
1395	hold_lkb(lkb);
1396	del_lkb(r, lkb);
1397	add_lkb(r, lkb, sts);
1398	unhold_lkb(lkb);
1399}
1400
1401static int msg_reply_type(int mstype)
1402{
1403	switch (mstype) {
1404	case DLM_MSG_REQUEST:
1405		return DLM_MSG_REQUEST_REPLY;
1406	case DLM_MSG_CONVERT:
1407		return DLM_MSG_CONVERT_REPLY;
1408	case DLM_MSG_UNLOCK:
1409		return DLM_MSG_UNLOCK_REPLY;
1410	case DLM_MSG_CANCEL:
1411		return DLM_MSG_CANCEL_REPLY;
1412	case DLM_MSG_LOOKUP:
1413		return DLM_MSG_LOOKUP_REPLY;
1414	}
1415	return -1;
1416}
1417
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1418/* add/remove lkb from global waiters list of lkb's waiting for
1419   a reply from a remote node */
1420
1421static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid)
1422{
1423	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1424	int error = 0;
1425
1426	mutex_lock(&ls->ls_waiters_mutex);
1427
1428	if (is_overlap_unlock(lkb) ||
1429	    (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) {
1430		error = -EINVAL;
1431		goto out;
1432	}
1433
1434	if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) {
1435		switch (mstype) {
1436		case DLM_MSG_UNLOCK:
1437			lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
1438			break;
1439		case DLM_MSG_CANCEL:
1440			lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
1441			break;
1442		default:
1443			error = -EBUSY;
1444			goto out;
1445		}
1446		lkb->lkb_wait_count++;
1447		hold_lkb(lkb);
1448
1449		log_debug(ls, "addwait %x cur %d overlap %d count %d f %x",
1450			  lkb->lkb_id, lkb->lkb_wait_type, mstype,
1451			  lkb->lkb_wait_count, lkb->lkb_flags);
1452		goto out;
1453	}
1454
1455	DLM_ASSERT(!lkb->lkb_wait_count,
1456		   dlm_print_lkb(lkb);
1457		   printk("wait_count %d\n", lkb->lkb_wait_count););
1458
1459	lkb->lkb_wait_count++;
1460	lkb->lkb_wait_type = mstype;
 
1461	lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */
1462	hold_lkb(lkb);
1463	list_add(&lkb->lkb_wait_reply, &ls->ls_waiters);
1464 out:
1465	if (error)
1466		log_error(ls, "addwait error %x %d flags %x %d %d %s",
1467			  lkb->lkb_id, error, lkb->lkb_flags, mstype,
1468			  lkb->lkb_wait_type, lkb->lkb_resource->res_name);
1469	mutex_unlock(&ls->ls_waiters_mutex);
1470	return error;
1471}
1472
1473/* We clear the RESEND flag because we might be taking an lkb off the waiters
1474   list as part of process_requestqueue (e.g. a lookup that has an optimized
1475   request reply on the requestqueue) between dlm_recover_waiters_pre() which
1476   set RESEND and dlm_recover_waiters_post() */
1477
1478static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype,
1479				struct dlm_message *ms)
1480{
1481	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1482	int overlap_done = 0;
1483
1484	if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) {
1485		log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id);
1486		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
1487		overlap_done = 1;
1488		goto out_del;
1489	}
1490
1491	if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) {
1492		log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id);
1493		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1494		overlap_done = 1;
1495		goto out_del;
1496	}
1497
1498	/* Cancel state was preemptively cleared by a successful convert,
1499	   see next comment, nothing to do. */
1500
1501	if ((mstype == DLM_MSG_CANCEL_REPLY) &&
1502	    (lkb->lkb_wait_type != DLM_MSG_CANCEL)) {
1503		log_debug(ls, "remwait %x cancel_reply wait_type %d",
1504			  lkb->lkb_id, lkb->lkb_wait_type);
1505		return -1;
1506	}
1507
1508	/* Remove for the convert reply, and premptively remove for the
1509	   cancel reply.  A convert has been granted while there's still
1510	   an outstanding cancel on it (the cancel is moot and the result
1511	   in the cancel reply should be 0).  We preempt the cancel reply
1512	   because the app gets the convert result and then can follow up
1513	   with another op, like convert.  This subsequent op would see the
1514	   lingering state of the cancel and fail with -EBUSY. */
1515
1516	if ((mstype == DLM_MSG_CONVERT_REPLY) &&
1517	    (lkb->lkb_wait_type == DLM_MSG_CONVERT) &&
1518	    is_overlap_cancel(lkb) && ms && !ms->m_result) {
1519		log_debug(ls, "remwait %x convert_reply zap overlap_cancel",
1520			  lkb->lkb_id);
1521		lkb->lkb_wait_type = 0;
1522		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
1523		lkb->lkb_wait_count--;
1524		unhold_lkb(lkb);
1525		goto out_del;
1526	}
1527
1528	/* N.B. type of reply may not always correspond to type of original
1529	   msg due to lookup->request optimization, verify others? */
1530
1531	if (lkb->lkb_wait_type) {
1532		lkb->lkb_wait_type = 0;
1533		goto out_del;
1534	}
1535
1536	log_error(ls, "remwait error %x remote %d %x msg %d flags %x no wait",
1537		  lkb->lkb_id, ms ? le32_to_cpu(ms->m_header.h_nodeid) : 0,
1538		  lkb->lkb_remid, mstype, lkb->lkb_flags);
1539	return -1;
1540
1541 out_del:
1542	/* the force-unlock/cancel has completed and we haven't recvd a reply
1543	   to the op that was in progress prior to the unlock/cancel; we
1544	   give up on any reply to the earlier op.  FIXME: not sure when/how
1545	   this would happen */
1546
1547	if (overlap_done && lkb->lkb_wait_type) {
1548		log_error(ls, "remwait error %x reply %d wait_type %d overlap",
1549			  lkb->lkb_id, mstype, lkb->lkb_wait_type);
1550		lkb->lkb_wait_count--;
1551		unhold_lkb(lkb);
1552		lkb->lkb_wait_type = 0;
1553	}
1554
1555	DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb););
1556
1557	lkb->lkb_flags &= ~DLM_IFL_RESEND;
1558	lkb->lkb_wait_count--;
1559	if (!lkb->lkb_wait_count)
1560		list_del_init(&lkb->lkb_wait_reply);
1561	unhold_lkb(lkb);
1562	return 0;
1563}
1564
1565static int remove_from_waiters(struct dlm_lkb *lkb, int mstype)
1566{
1567	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1568	int error;
1569
1570	mutex_lock(&ls->ls_waiters_mutex);
1571	error = _remove_from_waiters(lkb, mstype, NULL);
1572	mutex_unlock(&ls->ls_waiters_mutex);
1573	return error;
1574}
1575
1576/* Handles situations where we might be processing a "fake" or "stub" reply in
1577   which we can't try to take waiters_mutex again. */
1578
1579static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms)
1580{
1581	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1582	int error;
1583
1584	if (ms->m_flags != cpu_to_le32(DLM_IFL_STUB_MS))
1585		mutex_lock(&ls->ls_waiters_mutex);
1586	error = _remove_from_waiters(lkb, le32_to_cpu(ms->m_type), ms);
1587	if (ms->m_flags != cpu_to_le32(DLM_IFL_STUB_MS))
1588		mutex_unlock(&ls->ls_waiters_mutex);
1589	return error;
1590}
1591
1592static void shrink_bucket(struct dlm_ls *ls, int b)
1593{
1594	struct rb_node *n, *next;
1595	struct dlm_rsb *r;
1596	char *name;
1597	int our_nodeid = dlm_our_nodeid();
1598	int remote_count = 0;
1599	int need_shrink = 0;
1600	int i, len, rv;
1601
1602	memset(&ls->ls_remove_lens, 0, sizeof(int) * DLM_REMOVE_NAMES_MAX);
1603
1604	spin_lock(&ls->ls_rsbtbl[b].lock);
1605
1606	if (!(ls->ls_rsbtbl[b].flags & DLM_RTF_SHRINK)) {
1607		spin_unlock(&ls->ls_rsbtbl[b].lock);
1608		return;
1609	}
1610
1611	for (n = rb_first(&ls->ls_rsbtbl[b].toss); n; n = next) {
1612		next = rb_next(n);
1613		r = rb_entry(n, struct dlm_rsb, res_hashnode);
1614
1615		/* If we're the directory record for this rsb, and
1616		   we're not the master of it, then we need to wait
1617		   for the master node to send us a dir remove for
1618		   before removing the dir record. */
1619
1620		if (!dlm_no_directory(ls) &&
1621		    (r->res_master_nodeid != our_nodeid) &&
1622		    (dlm_dir_nodeid(r) == our_nodeid)) {
1623			continue;
1624		}
1625
1626		need_shrink = 1;
1627
1628		if (!time_after_eq(jiffies, r->res_toss_time +
1629				   dlm_config.ci_toss_secs * HZ)) {
1630			continue;
1631		}
1632
1633		if (!dlm_no_directory(ls) &&
1634		    (r->res_master_nodeid == our_nodeid) &&
1635		    (dlm_dir_nodeid(r) != our_nodeid)) {
1636
1637			/* We're the master of this rsb but we're not
1638			   the directory record, so we need to tell the
1639			   dir node to remove the dir record. */
1640
1641			ls->ls_remove_lens[remote_count] = r->res_length;
1642			memcpy(ls->ls_remove_names[remote_count], r->res_name,
1643			       DLM_RESNAME_MAXLEN);
1644			remote_count++;
1645
1646			if (remote_count >= DLM_REMOVE_NAMES_MAX)
1647				break;
1648			continue;
1649		}
1650
1651		if (!kref_put(&r->res_ref, kill_rsb)) {
1652			log_error(ls, "tossed rsb in use %s", r->res_name);
1653			continue;
1654		}
1655
1656		rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1657		dlm_free_rsb(r);
1658	}
1659
1660	if (need_shrink)
1661		ls->ls_rsbtbl[b].flags |= DLM_RTF_SHRINK;
1662	else
1663		ls->ls_rsbtbl[b].flags &= ~DLM_RTF_SHRINK;
1664	spin_unlock(&ls->ls_rsbtbl[b].lock);
 
1665
1666	/*
1667	 * While searching for rsb's to free, we found some that require
1668	 * remote removal.  We leave them in place and find them again here
1669	 * so there is a very small gap between removing them from the toss
1670	 * list and sending the removal.  Keeping this gap small is
1671	 * important to keep us (the master node) from being out of sync
1672	 * with the remote dir node for very long.
1673	 */
1674
1675	for (i = 0; i < remote_count; i++) {
1676		name = ls->ls_remove_names[i];
1677		len = ls->ls_remove_lens[i];
 
 
1678
 
 
1679		spin_lock(&ls->ls_rsbtbl[b].lock);
1680		rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
1681		if (rv) {
1682			spin_unlock(&ls->ls_rsbtbl[b].lock);
1683			log_debug(ls, "remove_name not toss %s", name);
1684			continue;
1685		}
1686
1687		if (r->res_master_nodeid != our_nodeid) {
1688			spin_unlock(&ls->ls_rsbtbl[b].lock);
1689			log_debug(ls, "remove_name master %d dir %d our %d %s",
1690				  r->res_master_nodeid, r->res_dir_nodeid,
1691				  our_nodeid, name);
1692			continue;
1693		}
1694
1695		if (r->res_dir_nodeid == our_nodeid) {
1696			/* should never happen */
1697			spin_unlock(&ls->ls_rsbtbl[b].lock);
1698			log_error(ls, "remove_name dir %d master %d our %d %s",
1699				  r->res_dir_nodeid, r->res_master_nodeid,
1700				  our_nodeid, name);
1701			continue;
1702		}
1703
1704		if (!time_after_eq(jiffies, r->res_toss_time +
1705				   dlm_config.ci_toss_secs * HZ)) {
1706			spin_unlock(&ls->ls_rsbtbl[b].lock);
1707			log_debug(ls, "remove_name toss_time %lu now %lu %s",
1708				  r->res_toss_time, jiffies, name);
1709			continue;
1710		}
1711
1712		if (!kref_put(&r->res_ref, kill_rsb)) {
 
 
 
 
1713			spin_unlock(&ls->ls_rsbtbl[b].lock);
1714			log_error(ls, "remove_name in use %s", name);
1715			continue;
1716		}
1717
1718		rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
1719		send_remove(r);
1720		spin_unlock(&ls->ls_rsbtbl[b].lock);
1721
1722		dlm_free_rsb(r);
1723	}
 
 
1724}
1725
1726void dlm_scan_rsbs(struct dlm_ls *ls)
1727{
1728	int i;
1729
1730	for (i = 0; i < ls->ls_rsbtbl_size; i++) {
1731		shrink_bucket(ls, i);
1732		if (dlm_locking_stopped(ls))
1733			break;
1734		cond_resched();
1735	}
1736}
1737
1738#ifdef CONFIG_DLM_DEPRECATED_API
1739static void add_timeout(struct dlm_lkb *lkb)
1740{
1741	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1742
1743	if (is_master_copy(lkb))
1744		return;
1745
1746	if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) &&
1747	    !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
1748		lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN;
1749		goto add_it;
1750	}
1751	if (lkb->lkb_exflags & DLM_LKF_TIMEOUT)
1752		goto add_it;
1753	return;
1754
1755 add_it:
1756	DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb););
1757	mutex_lock(&ls->ls_timeout_mutex);
1758	hold_lkb(lkb);
1759	list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout);
1760	mutex_unlock(&ls->ls_timeout_mutex);
1761}
1762
1763static void del_timeout(struct dlm_lkb *lkb)
1764{
1765	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
1766
1767	mutex_lock(&ls->ls_timeout_mutex);
1768	if (!list_empty(&lkb->lkb_time_list)) {
1769		list_del_init(&lkb->lkb_time_list);
1770		unhold_lkb(lkb);
1771	}
1772	mutex_unlock(&ls->ls_timeout_mutex);
1773}
1774
1775/* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and
1776   lkb_lksb_timeout without lock_rsb?  Note: we can't lock timeout_mutex
1777   and then lock rsb because of lock ordering in add_timeout.  We may need
1778   to specify some special timeout-related bits in the lkb that are just to
1779   be accessed under the timeout_mutex. */
1780
1781void dlm_scan_timeout(struct dlm_ls *ls)
1782{
1783	struct dlm_rsb *r;
1784	struct dlm_lkb *lkb = NULL, *iter;
1785	int do_cancel, do_warn;
1786	s64 wait_us;
1787
1788	for (;;) {
1789		if (dlm_locking_stopped(ls))
1790			break;
1791
1792		do_cancel = 0;
1793		do_warn = 0;
1794		mutex_lock(&ls->ls_timeout_mutex);
1795		list_for_each_entry(iter, &ls->ls_timeout, lkb_time_list) {
1796
1797			wait_us = ktime_to_us(ktime_sub(ktime_get(),
1798							iter->lkb_timestamp));
1799
1800			if ((iter->lkb_exflags & DLM_LKF_TIMEOUT) &&
1801			    wait_us >= (iter->lkb_timeout_cs * 10000))
1802				do_cancel = 1;
1803
1804			if ((iter->lkb_flags & DLM_IFL_WATCH_TIMEWARN) &&
1805			    wait_us >= dlm_config.ci_timewarn_cs * 10000)
1806				do_warn = 1;
1807
1808			if (!do_cancel && !do_warn)
1809				continue;
1810			hold_lkb(iter);
1811			lkb = iter;
1812			break;
1813		}
1814		mutex_unlock(&ls->ls_timeout_mutex);
1815
1816		if (!lkb)
1817			break;
1818
1819		r = lkb->lkb_resource;
1820		hold_rsb(r);
1821		lock_rsb(r);
1822
1823		if (do_warn) {
1824			/* clear flag so we only warn once */
1825			lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1826			if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT))
1827				del_timeout(lkb);
1828			dlm_timeout_warn(lkb);
1829		}
1830
1831		if (do_cancel) {
1832			log_debug(ls, "timeout cancel %x node %d %s",
1833				  lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
1834			lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN;
1835			lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL;
1836			del_timeout(lkb);
1837			_cancel_lock(r, lkb);
1838		}
1839
1840		unlock_rsb(r);
1841		unhold_rsb(r);
1842		dlm_put_lkb(lkb);
1843	}
1844}
1845
1846/* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping
1847   dlm_recoverd before checking/setting ls_recover_begin. */
1848
1849void dlm_adjust_timeouts(struct dlm_ls *ls)
1850{
1851	struct dlm_lkb *lkb;
1852	u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin);
1853
1854	ls->ls_recover_begin = 0;
1855	mutex_lock(&ls->ls_timeout_mutex);
1856	list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list)
1857		lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us);
1858	mutex_unlock(&ls->ls_timeout_mutex);
 
 
 
 
 
 
 
 
 
 
1859}
1860#else
1861static void add_timeout(struct dlm_lkb *lkb) { }
1862static void del_timeout(struct dlm_lkb *lkb) { }
1863#endif
1864
1865/* lkb is master or local copy */
1866
1867static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1868{
1869	int b, len = r->res_ls->ls_lvblen;
1870
1871	/* b=1 lvb returned to caller
1872	   b=0 lvb written to rsb or invalidated
1873	   b=-1 do nothing */
1874
1875	b =  dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1876
1877	if (b == 1) {
1878		if (!lkb->lkb_lvbptr)
1879			return;
1880
1881		if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1882			return;
1883
1884		if (!r->res_lvbptr)
1885			return;
1886
1887		memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len);
1888		lkb->lkb_lvbseq = r->res_lvbseq;
1889
1890	} else if (b == 0) {
1891		if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1892			rsb_set_flag(r, RSB_VALNOTVALID);
1893			return;
1894		}
1895
1896		if (!lkb->lkb_lvbptr)
1897			return;
1898
1899		if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1900			return;
1901
1902		if (!r->res_lvbptr)
1903			r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1904
1905		if (!r->res_lvbptr)
1906			return;
1907
1908		memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len);
1909		r->res_lvbseq++;
1910		lkb->lkb_lvbseq = r->res_lvbseq;
1911		rsb_clear_flag(r, RSB_VALNOTVALID);
1912	}
1913
1914	if (rsb_flag(r, RSB_VALNOTVALID))
1915		lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID;
1916}
1917
1918static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1919{
1920	if (lkb->lkb_grmode < DLM_LOCK_PW)
1921		return;
1922
1923	if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) {
1924		rsb_set_flag(r, RSB_VALNOTVALID);
1925		return;
1926	}
1927
1928	if (!lkb->lkb_lvbptr)
1929		return;
1930
1931	if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1932		return;
1933
1934	if (!r->res_lvbptr)
1935		r->res_lvbptr = dlm_allocate_lvb(r->res_ls);
1936
1937	if (!r->res_lvbptr)
1938		return;
1939
1940	memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
1941	r->res_lvbseq++;
1942	rsb_clear_flag(r, RSB_VALNOTVALID);
1943}
1944
1945/* lkb is process copy (pc) */
1946
1947static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
1948			    struct dlm_message *ms)
1949{
1950	int b;
1951
1952	if (!lkb->lkb_lvbptr)
1953		return;
1954
1955	if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
1956		return;
1957
1958	b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1];
1959	if (b == 1) {
1960		int len = receive_extralen(ms);
1961		if (len > r->res_ls->ls_lvblen)
1962			len = r->res_ls->ls_lvblen;
1963		memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
1964		lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
1965	}
1966}
1967
1968/* Manipulate lkb's on rsb's convert/granted/waiting queues
1969   remove_lock -- used for unlock, removes lkb from granted
1970   revert_lock -- used for cancel, moves lkb from convert to granted
1971   grant_lock  -- used for request and convert, adds lkb to granted or
1972                  moves lkb from convert or waiting to granted
1973
1974   Each of these is used for master or local copy lkb's.  There is
1975   also a _pc() variation used to make the corresponding change on
1976   a process copy (pc) lkb. */
1977
1978static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1979{
1980	del_lkb(r, lkb);
1981	lkb->lkb_grmode = DLM_LOCK_IV;
1982	/* this unhold undoes the original ref from create_lkb()
1983	   so this leads to the lkb being freed */
1984	unhold_lkb(lkb);
1985}
1986
1987static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
1988{
1989	set_lvb_unlock(r, lkb);
1990	_remove_lock(r, lkb);
1991}
1992
1993static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
1994{
1995	_remove_lock(r, lkb);
1996}
1997
1998/* returns: 0 did nothing
1999	    1 moved lock to granted
2000	   -1 removed lock */
2001
2002static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2003{
2004	int rv = 0;
2005
2006	lkb->lkb_rqmode = DLM_LOCK_IV;
2007
2008	switch (lkb->lkb_status) {
2009	case DLM_LKSTS_GRANTED:
2010		break;
2011	case DLM_LKSTS_CONVERT:
2012		move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2013		rv = 1;
2014		break;
2015	case DLM_LKSTS_WAITING:
2016		del_lkb(r, lkb);
2017		lkb->lkb_grmode = DLM_LOCK_IV;
2018		/* this unhold undoes the original ref from create_lkb()
2019		   so this leads to the lkb being freed */
2020		unhold_lkb(lkb);
2021		rv = -1;
2022		break;
2023	default:
2024		log_print("invalid status for revert %d", lkb->lkb_status);
2025	}
2026	return rv;
2027}
2028
2029static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb)
2030{
2031	return revert_lock(r, lkb);
2032}
2033
2034static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2035{
2036	if (lkb->lkb_grmode != lkb->lkb_rqmode) {
2037		lkb->lkb_grmode = lkb->lkb_rqmode;
2038		if (lkb->lkb_status)
2039			move_lkb(r, lkb, DLM_LKSTS_GRANTED);
2040		else
2041			add_lkb(r, lkb, DLM_LKSTS_GRANTED);
2042	}
2043
2044	lkb->lkb_rqmode = DLM_LOCK_IV;
2045	lkb->lkb_highbast = 0;
2046}
2047
2048static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
2049{
2050	set_lvb_lock(r, lkb);
2051	_grant_lock(r, lkb);
2052}
2053
2054static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb,
2055			  struct dlm_message *ms)
2056{
2057	set_lvb_lock_pc(r, lkb, ms);
2058	_grant_lock(r, lkb);
2059}
2060
2061/* called by grant_pending_locks() which means an async grant message must
2062   be sent to the requesting node in addition to granting the lock if the
2063   lkb belongs to a remote node. */
2064
2065static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb)
2066{
2067	grant_lock(r, lkb);
2068	if (is_master_copy(lkb))
2069		send_grant(r, lkb);
2070	else
2071		queue_cast(r, lkb, 0);
2072}
2073
2074/* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to
2075   change the granted/requested modes.  We're munging things accordingly in
2076   the process copy.
2077   CONVDEADLK: our grmode may have been forced down to NL to resolve a
2078   conversion deadlock
2079   ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become
2080   compatible with other granted locks */
2081
2082static void munge_demoted(struct dlm_lkb *lkb)
2083{
2084	if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) {
2085		log_print("munge_demoted %x invalid modes gr %d rq %d",
2086			  lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode);
2087		return;
2088	}
2089
2090	lkb->lkb_grmode = DLM_LOCK_NL;
2091}
2092
2093static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms)
2094{
2095	if (ms->m_type != cpu_to_le32(DLM_MSG_REQUEST_REPLY) &&
2096	    ms->m_type != cpu_to_le32(DLM_MSG_GRANT)) {
2097		log_print("munge_altmode %x invalid reply type %d",
2098			  lkb->lkb_id, le32_to_cpu(ms->m_type));
2099		return;
2100	}
2101
2102	if (lkb->lkb_exflags & DLM_LKF_ALTPR)
2103		lkb->lkb_rqmode = DLM_LOCK_PR;
2104	else if (lkb->lkb_exflags & DLM_LKF_ALTCW)
2105		lkb->lkb_rqmode = DLM_LOCK_CW;
2106	else {
2107		log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags);
2108		dlm_print_lkb(lkb);
2109	}
2110}
2111
2112static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head)
2113{
2114	struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb,
2115					   lkb_statequeue);
2116	if (lkb->lkb_id == first->lkb_id)
2117		return 1;
2118
2119	return 0;
2120}
2121
2122/* Check if the given lkb conflicts with another lkb on the queue. */
2123
2124static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb)
2125{
2126	struct dlm_lkb *this;
2127
2128	list_for_each_entry(this, head, lkb_statequeue) {
2129		if (this == lkb)
2130			continue;
2131		if (!modes_compat(this, lkb))
2132			return 1;
2133	}
2134	return 0;
2135}
2136
2137/*
2138 * "A conversion deadlock arises with a pair of lock requests in the converting
2139 * queue for one resource.  The granted mode of each lock blocks the requested
2140 * mode of the other lock."
2141 *
2142 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the
2143 * convert queue from being granted, then deadlk/demote lkb.
2144 *
2145 * Example:
2146 * Granted Queue: empty
2147 * Convert Queue: NL->EX (first lock)
2148 *                PR->EX (second lock)
2149 *
2150 * The first lock can't be granted because of the granted mode of the second
2151 * lock and the second lock can't be granted because it's not first in the
2152 * list.  We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we
2153 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK
2154 * flag set and return DEMOTED in the lksb flags.
2155 *
2156 * Originally, this function detected conv-deadlk in a more limited scope:
2157 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or
2158 * - if lkb1 was the first entry in the queue (not just earlier), and was
2159 *   blocked by the granted mode of lkb2, and there was nothing on the
2160 *   granted queue preventing lkb1 from being granted immediately, i.e.
2161 *   lkb2 was the only thing preventing lkb1 from being granted.
2162 *
2163 * That second condition meant we'd only say there was conv-deadlk if
2164 * resolving it (by demotion) would lead to the first lock on the convert
2165 * queue being granted right away.  It allowed conversion deadlocks to exist
2166 * between locks on the convert queue while they couldn't be granted anyway.
2167 *
2168 * Now, we detect and take action on conversion deadlocks immediately when
2169 * they're created, even if they may not be immediately consequential.  If
2170 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted
2171 * mode that would prevent lkb1's conversion from being granted, we do a
2172 * deadlk/demote on lkb2 right away and don't let it onto the convert queue.
2173 * I think this means that the lkb_is_ahead condition below should always
2174 * be zero, i.e. there will never be conv-deadlk between two locks that are
2175 * both already on the convert queue.
2176 */
2177
2178static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2)
2179{
2180	struct dlm_lkb *lkb1;
2181	int lkb_is_ahead = 0;
2182
2183	list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) {
2184		if (lkb1 == lkb2) {
2185			lkb_is_ahead = 1;
2186			continue;
2187		}
2188
2189		if (!lkb_is_ahead) {
2190			if (!modes_compat(lkb2, lkb1))
2191				return 1;
2192		} else {
2193			if (!modes_compat(lkb2, lkb1) &&
2194			    !modes_compat(lkb1, lkb2))
2195				return 1;
2196		}
2197	}
2198	return 0;
2199}
2200
2201/*
2202 * Return 1 if the lock can be granted, 0 otherwise.
2203 * Also detect and resolve conversion deadlocks.
2204 *
2205 * lkb is the lock to be granted
2206 *
2207 * now is 1 if the function is being called in the context of the
2208 * immediate request, it is 0 if called later, after the lock has been
2209 * queued.
2210 *
2211 * recover is 1 if dlm_recover_grant() is trying to grant conversions
2212 * after recovery.
2213 *
2214 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis
2215 */
2216
2217static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2218			   int recover)
2219{
2220	int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV);
2221
2222	/*
2223	 * 6-10: Version 5.4 introduced an option to address the phenomenon of
2224	 * a new request for a NL mode lock being blocked.
2225	 *
2226	 * 6-11: If the optional EXPEDITE flag is used with the new NL mode
2227	 * request, then it would be granted.  In essence, the use of this flag
2228	 * tells the Lock Manager to expedite theis request by not considering
2229	 * what may be in the CONVERTING or WAITING queues...  As of this
2230	 * writing, the EXPEDITE flag can be used only with new requests for NL
2231	 * mode locks.  This flag is not valid for conversion requests.
2232	 *
2233	 * A shortcut.  Earlier checks return an error if EXPEDITE is used in a
2234	 * conversion or used with a non-NL requested mode.  We also know an
2235	 * EXPEDITE request is always granted immediately, so now must always
2236	 * be 1.  The full condition to grant an expedite request: (now &&
2237	 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can
2238	 * therefore be shortened to just checking the flag.
2239	 */
2240
2241	if (lkb->lkb_exflags & DLM_LKF_EXPEDITE)
2242		return 1;
2243
2244	/*
2245	 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be
2246	 * added to the remaining conditions.
2247	 */
2248
2249	if (queue_conflict(&r->res_grantqueue, lkb))
2250		return 0;
2251
2252	/*
2253	 * 6-3: By default, a conversion request is immediately granted if the
2254	 * requested mode is compatible with the modes of all other granted
2255	 * locks
2256	 */
2257
2258	if (queue_conflict(&r->res_convertqueue, lkb))
2259		return 0;
2260
2261	/*
2262	 * The RECOVER_GRANT flag means dlm_recover_grant() is granting
2263	 * locks for a recovered rsb, on which lkb's have been rebuilt.
2264	 * The lkb's may have been rebuilt on the queues in a different
2265	 * order than they were in on the previous master.  So, granting
2266	 * queued conversions in order after recovery doesn't make sense
2267	 * since the order hasn't been preserved anyway.  The new order
2268	 * could also have created a new "in place" conversion deadlock.
2269	 * (e.g. old, failed master held granted EX, with PR->EX, NL->EX.
2270	 * After recovery, there would be no granted locks, and possibly
2271	 * NL->EX, PR->EX, an in-place conversion deadlock.)  So, after
2272	 * recovery, grant conversions without considering order.
2273	 */
2274
2275	if (conv && recover)
2276		return 1;
2277
2278	/*
2279	 * 6-5: But the default algorithm for deciding whether to grant or
2280	 * queue conversion requests does not by itself guarantee that such
2281	 * requests are serviced on a "first come first serve" basis.  This, in
2282	 * turn, can lead to a phenomenon known as "indefinate postponement".
2283	 *
2284	 * 6-7: This issue is dealt with by using the optional QUECVT flag with
2285	 * the system service employed to request a lock conversion.  This flag
2286	 * forces certain conversion requests to be queued, even if they are
2287	 * compatible with the granted modes of other locks on the same
2288	 * resource.  Thus, the use of this flag results in conversion requests
2289	 * being ordered on a "first come first servce" basis.
2290	 *
2291	 * DCT: This condition is all about new conversions being able to occur
2292	 * "in place" while the lock remains on the granted queue (assuming
2293	 * nothing else conflicts.)  IOW if QUECVT isn't set, a conversion
2294	 * doesn't _have_ to go onto the convert queue where it's processed in
2295	 * order.  The "now" variable is necessary to distinguish converts
2296	 * being received and processed for the first time now, because once a
2297	 * convert is moved to the conversion queue the condition below applies
2298	 * requiring fifo granting.
2299	 */
2300
2301	if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT))
2302		return 1;
2303
2304	/*
2305	 * Even if the convert is compat with all granted locks,
2306	 * QUECVT forces it behind other locks on the convert queue.
2307	 */
2308
2309	if (now && conv && (lkb->lkb_exflags & DLM_LKF_QUECVT)) {
2310		if (list_empty(&r->res_convertqueue))
2311			return 1;
2312		else
2313			return 0;
2314	}
2315
2316	/*
2317	 * The NOORDER flag is set to avoid the standard vms rules on grant
2318	 * order.
2319	 */
2320
2321	if (lkb->lkb_exflags & DLM_LKF_NOORDER)
2322		return 1;
2323
2324	/*
2325	 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be
2326	 * granted until all other conversion requests ahead of it are granted
2327	 * and/or canceled.
2328	 */
2329
2330	if (!now && conv && first_in_list(lkb, &r->res_convertqueue))
2331		return 1;
2332
2333	/*
2334	 * 6-4: By default, a new request is immediately granted only if all
2335	 * three of the following conditions are satisfied when the request is
2336	 * issued:
2337	 * - The queue of ungranted conversion requests for the resource is
2338	 *   empty.
2339	 * - The queue of ungranted new requests for the resource is empty.
2340	 * - The mode of the new request is compatible with the most
2341	 *   restrictive mode of all granted locks on the resource.
2342	 */
2343
2344	if (now && !conv && list_empty(&r->res_convertqueue) &&
2345	    list_empty(&r->res_waitqueue))
2346		return 1;
2347
2348	/*
2349	 * 6-4: Once a lock request is in the queue of ungranted new requests,
2350	 * it cannot be granted until the queue of ungranted conversion
2351	 * requests is empty, all ungranted new requests ahead of it are
2352	 * granted and/or canceled, and it is compatible with the granted mode
2353	 * of the most restrictive lock granted on the resource.
2354	 */
2355
2356	if (!now && !conv && list_empty(&r->res_convertqueue) &&
2357	    first_in_list(lkb, &r->res_waitqueue))
2358		return 1;
2359
2360	return 0;
2361}
2362
2363static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
2364			  int recover, int *err)
2365{
2366	int rv;
2367	int8_t alt = 0, rqmode = lkb->lkb_rqmode;
2368	int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV);
2369
2370	if (err)
2371		*err = 0;
2372
2373	rv = _can_be_granted(r, lkb, now, recover);
2374	if (rv)
2375		goto out;
2376
2377	/*
2378	 * The CONVDEADLK flag is non-standard and tells the dlm to resolve
2379	 * conversion deadlocks by demoting grmode to NL, otherwise the dlm
2380	 * cancels one of the locks.
2381	 */
2382
2383	if (is_convert && can_be_queued(lkb) &&
2384	    conversion_deadlock_detect(r, lkb)) {
2385		if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) {
2386			lkb->lkb_grmode = DLM_LOCK_NL;
2387			lkb->lkb_sbflags |= DLM_SBF_DEMOTED;
2388		} else if (err) {
2389			*err = -EDEADLK;
2390		} else {
2391			log_print("can_be_granted deadlock %x now %d",
2392				  lkb->lkb_id, now);
2393			dlm_dump_rsb(r);
 
 
2394		}
2395		goto out;
2396	}
2397
2398	/*
2399	 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try
2400	 * to grant a request in a mode other than the normal rqmode.  It's a
2401	 * simple way to provide a big optimization to applications that can
2402	 * use them.
2403	 */
2404
2405	if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR))
2406		alt = DLM_LOCK_PR;
2407	else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW))
2408		alt = DLM_LOCK_CW;
2409
2410	if (alt) {
2411		lkb->lkb_rqmode = alt;
2412		rv = _can_be_granted(r, lkb, now, 0);
2413		if (rv)
2414			lkb->lkb_sbflags |= DLM_SBF_ALTMODE;
2415		else
2416			lkb->lkb_rqmode = rqmode;
2417	}
2418 out:
2419	return rv;
2420}
2421
 
 
 
 
 
 
 
2422/* Returns the highest requested mode of all blocked conversions; sets
2423   cw if there's a blocked conversion to DLM_LOCK_CW. */
2424
2425static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw,
2426				 unsigned int *count)
2427{
2428	struct dlm_lkb *lkb, *s;
2429	int recover = rsb_flag(r, RSB_RECOVER_GRANT);
2430	int hi, demoted, quit, grant_restart, demote_restart;
2431	int deadlk;
2432
2433	quit = 0;
2434 restart:
2435	grant_restart = 0;
2436	demote_restart = 0;
2437	hi = DLM_LOCK_IV;
2438
2439	list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) {
2440		demoted = is_demoted(lkb);
2441		deadlk = 0;
2442
2443		if (can_be_granted(r, lkb, 0, recover, &deadlk)) {
2444			grant_lock_pending(r, lkb);
2445			grant_restart = 1;
2446			if (count)
2447				(*count)++;
2448			continue;
2449		}
2450
2451		if (!demoted && is_demoted(lkb)) {
2452			log_print("WARN: pending demoted %x node %d %s",
2453				  lkb->lkb_id, lkb->lkb_nodeid, r->res_name);
2454			demote_restart = 1;
2455			continue;
2456		}
2457
2458		if (deadlk) {
2459			/*
2460			 * If DLM_LKB_NODLKWT flag is set and conversion
2461			 * deadlock is detected, we request blocking AST and
2462			 * down (or cancel) conversion.
2463			 */
2464			if (lkb->lkb_exflags & DLM_LKF_NODLCKWT) {
2465				if (lkb->lkb_highbast < lkb->lkb_rqmode) {
2466					queue_bast(r, lkb, lkb->lkb_rqmode);
2467					lkb->lkb_highbast = lkb->lkb_rqmode;
2468				}
2469			} else {
2470				log_print("WARN: pending deadlock %x node %d %s",
2471					  lkb->lkb_id, lkb->lkb_nodeid,
2472					  r->res_name);
2473				dlm_dump_rsb(r);
2474			}
2475			continue;
2476		}
2477
2478		hi = max_t(int, lkb->lkb_rqmode, hi);
2479
2480		if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
2481			*cw = 1;
2482	}
2483
2484	if (grant_restart)
2485		goto restart;
2486	if (demote_restart && !quit) {
2487		quit = 1;
2488		goto restart;
2489	}
2490
2491	return max_t(int, high, hi);
2492}
2493
2494static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw,
2495			      unsigned int *count)
2496{
2497	struct dlm_lkb *lkb, *s;
2498
2499	list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
2500		if (can_be_granted(r, lkb, 0, 0, NULL)) {
2501			grant_lock_pending(r, lkb);
2502			if (count)
2503				(*count)++;
2504		} else {
2505			high = max_t(int, lkb->lkb_rqmode, high);
2506			if (lkb->lkb_rqmode == DLM_LOCK_CW)
2507				*cw = 1;
2508		}
2509	}
2510
2511	return high;
2512}
2513
2514/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
2515   on either the convert or waiting queue.
2516   high is the largest rqmode of all locks blocked on the convert or
2517   waiting queue. */
2518
2519static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
2520{
2521	if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
2522		if (gr->lkb_highbast < DLM_LOCK_EX)
2523			return 1;
2524		return 0;
2525	}
2526
2527	if (gr->lkb_highbast < high &&
2528	    !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
2529		return 1;
2530	return 0;
2531}
2532
2533static void grant_pending_locks(struct dlm_rsb *r, unsigned int *count)
2534{
2535	struct dlm_lkb *lkb, *s;
2536	int high = DLM_LOCK_IV;
2537	int cw = 0;
2538
2539	if (!is_master(r)) {
2540		log_print("grant_pending_locks r nodeid %d", r->res_nodeid);
2541		dlm_dump_rsb(r);
2542		return;
2543	}
2544
2545	high = grant_pending_convert(r, high, &cw, count);
2546	high = grant_pending_wait(r, high, &cw, count);
2547
2548	if (high == DLM_LOCK_IV)
2549		return;
2550
2551	/*
2552	 * If there are locks left on the wait/convert queue then send blocking
2553	 * ASTs to granted locks based on the largest requested mode (high)
2554	 * found above.
2555	 */
2556
2557	list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
2558		if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) {
2559			if (cw && high == DLM_LOCK_PR &&
2560			    lkb->lkb_grmode == DLM_LOCK_PR)
2561				queue_bast(r, lkb, DLM_LOCK_CW);
2562			else
2563				queue_bast(r, lkb, high);
2564			lkb->lkb_highbast = high;
2565		}
2566	}
2567}
2568
2569static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
2570{
2571	if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
2572	    (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
2573		if (gr->lkb_highbast < DLM_LOCK_EX)
2574			return 1;
2575		return 0;
2576	}
2577
2578	if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
2579		return 1;
2580	return 0;
2581}
2582
2583static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
2584			    struct dlm_lkb *lkb)
2585{
2586	struct dlm_lkb *gr;
2587
2588	list_for_each_entry(gr, head, lkb_statequeue) {
2589		/* skip self when sending basts to convertqueue */
2590		if (gr == lkb)
2591			continue;
2592		if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) {
2593			queue_bast(r, gr, lkb->lkb_rqmode);
2594			gr->lkb_highbast = lkb->lkb_rqmode;
2595		}
2596	}
2597}
2598
2599static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb)
2600{
2601	send_bast_queue(r, &r->res_grantqueue, lkb);
2602}
2603
2604static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb)
2605{
2606	send_bast_queue(r, &r->res_grantqueue, lkb);
2607	send_bast_queue(r, &r->res_convertqueue, lkb);
2608}
2609
2610/* set_master(r, lkb) -- set the master nodeid of a resource
2611
2612   The purpose of this function is to set the nodeid field in the given
2613   lkb using the nodeid field in the given rsb.  If the rsb's nodeid is
2614   known, it can just be copied to the lkb and the function will return
2615   0.  If the rsb's nodeid is _not_ known, it needs to be looked up
2616   before it can be copied to the lkb.
2617
2618   When the rsb nodeid is being looked up remotely, the initial lkb
2619   causing the lookup is kept on the ls_waiters list waiting for the
2620   lookup reply.  Other lkb's waiting for the same rsb lookup are kept
2621   on the rsb's res_lookup list until the master is verified.
2622
2623   Return values:
2624   0: nodeid is set in rsb/lkb and the caller should go ahead and use it
2625   1: the rsb master is not available and the lkb has been placed on
2626      a wait queue
2627*/
2628
2629static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb)
2630{
2631	int our_nodeid = dlm_our_nodeid();
 
2632
2633	if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) {
2634		rsb_clear_flag(r, RSB_MASTER_UNCERTAIN);
2635		r->res_first_lkid = lkb->lkb_id;
2636		lkb->lkb_nodeid = r->res_nodeid;
2637		return 0;
2638	}
2639
2640	if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) {
2641		list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup);
2642		return 1;
2643	}
2644
2645	if (r->res_master_nodeid == our_nodeid) {
2646		lkb->lkb_nodeid = 0;
2647		return 0;
2648	}
2649
2650	if (r->res_master_nodeid) {
2651		lkb->lkb_nodeid = r->res_master_nodeid;
2652		return 0;
2653	}
2654
2655	if (dlm_dir_nodeid(r) == our_nodeid) {
2656		/* This is a somewhat unusual case; find_rsb will usually
2657		   have set res_master_nodeid when dir nodeid is local, but
2658		   there are cases where we become the dir node after we've
2659		   past find_rsb and go through _request_lock again.
2660		   confirm_master() or process_lookup_list() needs to be
2661		   called after this. */
2662		log_debug(r->res_ls, "set_master %x self master %d dir %d %s",
2663			  lkb->lkb_id, r->res_master_nodeid, r->res_dir_nodeid,
2664			  r->res_name);
2665		r->res_master_nodeid = our_nodeid;
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
2666		r->res_nodeid = 0;
2667		lkb->lkb_nodeid = 0;
2668		return 0;
 
 
 
2669	}
2670
2671	r->res_first_lkid = lkb->lkb_id;
2672	send_lookup(r, lkb);
2673	return 1;
2674}
2675
2676static void process_lookup_list(struct dlm_rsb *r)
2677{
2678	struct dlm_lkb *lkb, *safe;
2679
2680	list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) {
2681		list_del_init(&lkb->lkb_rsb_lookup);
2682		_request_lock(r, lkb);
2683		schedule();
2684	}
2685}
2686
2687/* confirm_master -- confirm (or deny) an rsb's master nodeid */
2688
2689static void confirm_master(struct dlm_rsb *r, int error)
2690{
2691	struct dlm_lkb *lkb;
2692
2693	if (!r->res_first_lkid)
2694		return;
2695
2696	switch (error) {
2697	case 0:
2698	case -EINPROGRESS:
2699		r->res_first_lkid = 0;
2700		process_lookup_list(r);
2701		break;
2702
2703	case -EAGAIN:
2704	case -EBADR:
2705	case -ENOTBLK:
2706		/* the remote request failed and won't be retried (it was
2707		   a NOQUEUE, or has been canceled/unlocked); make a waiting
2708		   lkb the first_lkid */
2709
2710		r->res_first_lkid = 0;
2711
2712		if (!list_empty(&r->res_lookup)) {
2713			lkb = list_entry(r->res_lookup.next, struct dlm_lkb,
2714					 lkb_rsb_lookup);
2715			list_del_init(&lkb->lkb_rsb_lookup);
2716			r->res_first_lkid = lkb->lkb_id;
2717			_request_lock(r, lkb);
2718		}
2719		break;
2720
2721	default:
2722		log_error(r->res_ls, "confirm_master unknown error %d", error);
2723	}
2724}
2725
2726#ifdef CONFIG_DLM_DEPRECATED_API
2727static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2728			 int namelen, unsigned long timeout_cs,
2729			 void (*ast) (void *astparam),
2730			 void *astparam,
2731			 void (*bast) (void *astparam, int mode),
2732			 struct dlm_args *args)
2733#else
2734static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags,
2735			 int namelen, void (*ast)(void *astparam),
2736			 void *astparam,
2737			 void (*bast)(void *astparam, int mode),
2738			 struct dlm_args *args)
2739#endif
2740{
2741	int rv = -EINVAL;
2742
2743	/* check for invalid arg usage */
2744
2745	if (mode < 0 || mode > DLM_LOCK_EX)
2746		goto out;
2747
2748	if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN))
2749		goto out;
2750
2751	if (flags & DLM_LKF_CANCEL)
2752		goto out;
2753
2754	if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT))
2755		goto out;
2756
2757	if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT))
2758		goto out;
2759
2760	if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE)
2761		goto out;
2762
2763	if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT)
2764		goto out;
2765
2766	if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT)
2767		goto out;
2768
2769	if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE)
2770		goto out;
2771
2772	if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL)
2773		goto out;
2774
2775	if (!ast || !lksb)
2776		goto out;
2777
2778	if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr)
2779		goto out;
2780
2781	if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid)
2782		goto out;
2783
2784	/* these args will be copied to the lkb in validate_lock_args,
2785	   it cannot be done now because when converting locks, fields in
2786	   an active lkb cannot be modified before locking the rsb */
2787
2788	args->flags = flags;
2789	args->astfn = ast;
2790	args->astparam = astparam;
2791	args->bastfn = bast;
2792#ifdef CONFIG_DLM_DEPRECATED_API
2793	args->timeout = timeout_cs;
2794#endif
2795	args->mode = mode;
2796	args->lksb = lksb;
2797	rv = 0;
2798 out:
2799	return rv;
2800}
2801
2802static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args)
2803{
2804	if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK |
2805 		      DLM_LKF_FORCEUNLOCK))
2806		return -EINVAL;
2807
2808	if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK)
2809		return -EINVAL;
2810
2811	args->flags = flags;
2812	args->astparam = astarg;
2813	return 0;
2814}
2815
2816static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
2817			      struct dlm_args *args)
2818{
2819	int rv = -EBUSY;
2820
2821	if (args->flags & DLM_LKF_CONVERT) {
2822		if (lkb->lkb_status != DLM_LKSTS_GRANTED)
2823			goto out;
2824
2825		/* lock not allowed if there's any op in progress */
2826		if (lkb->lkb_wait_type || lkb->lkb_wait_count)
2827			goto out;
2828
2829		if (is_overlap(lkb))
 
2830			goto out;
2831
2832		rv = -EINVAL;
2833		if (lkb->lkb_flags & DLM_IFL_MSTCPY)
2834			goto out;
2835
2836		if (args->flags & DLM_LKF_QUECVT &&
2837		    !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1])
2838			goto out;
2839	}
2840
2841	lkb->lkb_exflags = args->flags;
2842	lkb->lkb_sbflags = 0;
2843	lkb->lkb_astfn = args->astfn;
2844	lkb->lkb_astparam = args->astparam;
2845	lkb->lkb_bastfn = args->bastfn;
2846	lkb->lkb_rqmode = args->mode;
2847	lkb->lkb_lksb = args->lksb;
2848	lkb->lkb_lvbptr = args->lksb->sb_lvbptr;
2849	lkb->lkb_ownpid = (int) current->pid;
2850#ifdef CONFIG_DLM_DEPRECATED_API
2851	lkb->lkb_timeout_cs = args->timeout;
2852#endif
2853	rv = 0;
2854 out:
2855	switch (rv) {
2856	case 0:
2857		break;
2858	case -EINVAL:
2859		/* annoy the user because dlm usage is wrong */
2860		WARN_ON(1);
2861		log_error(ls, "%s %d %x %x %x %d %d %s", __func__,
2862			  rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2863			  lkb->lkb_status, lkb->lkb_wait_type,
2864			  lkb->lkb_resource->res_name);
2865		break;
2866	default:
2867		log_debug(ls, "%s %d %x %x %x %d %d %s", __func__,
2868			  rv, lkb->lkb_id, lkb->lkb_flags, args->flags,
2869			  lkb->lkb_status, lkb->lkb_wait_type,
2870			  lkb->lkb_resource->res_name);
2871		break;
2872	}
2873
2874	return rv;
2875}
2876
2877/* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0
2878   for success */
2879
2880/* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here
2881   because there may be a lookup in progress and it's valid to do
2882   cancel/unlockf on it */
2883
2884static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args)
2885{
2886	struct dlm_ls *ls = lkb->lkb_resource->res_ls;
2887	int rv = -EBUSY;
2888
2889	/* normal unlock not allowed if there's any op in progress */
2890	if (!(args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) &&
2891	    (lkb->lkb_wait_type || lkb->lkb_wait_count))
2892		goto out;
 
 
 
 
 
 
 
 
 
 
 
2893
2894	/* an lkb may be waiting for an rsb lookup to complete where the
2895	   lookup was initiated by another lock */
2896
2897	if (!list_empty(&lkb->lkb_rsb_lookup)) {
2898		if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) {
2899			log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id);
2900			list_del_init(&lkb->lkb_rsb_lookup);
2901			queue_cast(lkb->lkb_resource, lkb,
2902				   args->flags & DLM_LKF_CANCEL ?
2903				   -DLM_ECANCEL : -DLM_EUNLOCK);
2904			unhold_lkb(lkb); /* undoes create_lkb() */
2905		}
2906		/* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */
2907		goto out;
2908	}
2909
2910	rv = -EINVAL;
2911	if (lkb->lkb_flags & DLM_IFL_MSTCPY) {
2912		log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id);
2913		dlm_print_lkb(lkb);
2914		goto out;
2915	}
2916
2917	/* an lkb may still exist even though the lock is EOL'ed due to a
2918	 * cancel, unlock or failed noqueue request; an app can't use these
2919	 * locks; return same error as if the lkid had not been found at all
2920	 */
2921
2922	if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) {
2923		log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id);
2924		rv = -ENOENT;
2925		goto out;
2926	}
2927
2928	/* cancel not allowed with another cancel/unlock in progress */
2929
2930	if (args->flags & DLM_LKF_CANCEL) {
2931		if (lkb->lkb_exflags & DLM_LKF_CANCEL)
2932			goto out;
2933
2934		if (is_overlap(lkb))
2935			goto out;
2936
2937		/* don't let scand try to do a cancel */
2938		del_timeout(lkb);
2939
2940		if (lkb->lkb_flags & DLM_IFL_RESEND) {
2941			lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2942			rv = -EBUSY;
2943			goto out;
2944		}
2945
2946		/* there's nothing to cancel */
2947		if (lkb->lkb_status == DLM_LKSTS_GRANTED &&
2948		    !lkb->lkb_wait_type) {
2949			rv = -EBUSY;
2950			goto out;
2951		}
2952
2953		switch (lkb->lkb_wait_type) {
2954		case DLM_MSG_LOOKUP:
2955		case DLM_MSG_REQUEST:
2956			lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL;
2957			rv = -EBUSY;
2958			goto out;
2959		case DLM_MSG_UNLOCK:
2960		case DLM_MSG_CANCEL:
2961			goto out;
2962		}
2963		/* add_to_waiters() will set OVERLAP_CANCEL */
2964		goto out_ok;
2965	}
2966
2967	/* do we need to allow a force-unlock if there's a normal unlock
2968	   already in progress?  in what conditions could the normal unlock
2969	   fail such that we'd want to send a force-unlock to be sure? */
2970
2971	if (args->flags & DLM_LKF_FORCEUNLOCK) {
2972		if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK)
2973			goto out;
2974
2975		if (is_overlap_unlock(lkb))
2976			goto out;
2977
2978		/* don't let scand try to do a cancel */
2979		del_timeout(lkb);
2980
2981		if (lkb->lkb_flags & DLM_IFL_RESEND) {
2982			lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2983			rv = -EBUSY;
2984			goto out;
2985		}
2986
2987		switch (lkb->lkb_wait_type) {
2988		case DLM_MSG_LOOKUP:
2989		case DLM_MSG_REQUEST:
2990			lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK;
2991			rv = -EBUSY;
2992			goto out;
2993		case DLM_MSG_UNLOCK:
2994			goto out;
2995		}
2996		/* add_to_waiters() will set OVERLAP_UNLOCK */
 
2997	}
2998
 
 
 
 
 
2999 out_ok:
3000	/* an overlapping op shouldn't blow away exflags from other op */
3001	lkb->lkb_exflags |= args->flags;
3002	lkb->lkb_sbflags = 0;
3003	lkb->lkb_astparam = args->astparam;
3004	rv = 0;
3005 out:
3006	switch (rv) {
3007	case 0:
3008		break;
3009	case -EINVAL:
3010		/* annoy the user because dlm usage is wrong */
3011		WARN_ON(1);
3012		log_error(ls, "%s %d %x %x %x %x %d %s", __func__, rv,
3013			  lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3014			  args->flags, lkb->lkb_wait_type,
3015			  lkb->lkb_resource->res_name);
3016		break;
3017	default:
3018		log_debug(ls, "%s %d %x %x %x %x %d %s", __func__, rv,
3019			  lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags,
3020			  args->flags, lkb->lkb_wait_type,
3021			  lkb->lkb_resource->res_name);
3022		break;
3023	}
3024
3025	return rv;
3026}
3027
3028/*
3029 * Four stage 4 varieties:
3030 * do_request(), do_convert(), do_unlock(), do_cancel()
3031 * These are called on the master node for the given lock and
3032 * from the central locking logic.
3033 */
3034
3035static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3036{
3037	int error = 0;
3038
3039	if (can_be_granted(r, lkb, 1, 0, NULL)) {
3040		grant_lock(r, lkb);
3041		queue_cast(r, lkb, 0);
3042		goto out;
3043	}
3044
3045	if (can_be_queued(lkb)) {
3046		error = -EINPROGRESS;
3047		add_lkb(r, lkb, DLM_LKSTS_WAITING);
3048		add_timeout(lkb);
3049		goto out;
3050	}
3051
3052	error = -EAGAIN;
3053	queue_cast(r, lkb, -EAGAIN);
3054 out:
3055	return error;
3056}
3057
3058static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3059			       int error)
3060{
3061	switch (error) {
3062	case -EAGAIN:
3063		if (force_blocking_asts(lkb))
3064			send_blocking_asts_all(r, lkb);
3065		break;
3066	case -EINPROGRESS:
3067		send_blocking_asts(r, lkb);
3068		break;
3069	}
3070}
3071
3072static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3073{
3074	int error = 0;
3075	int deadlk = 0;
3076
3077	/* changing an existing lock may allow others to be granted */
3078
3079	if (can_be_granted(r, lkb, 1, 0, &deadlk)) {
3080		grant_lock(r, lkb);
3081		queue_cast(r, lkb, 0);
3082		goto out;
3083	}
3084
3085	/* can_be_granted() detected that this lock would block in a conversion
3086	   deadlock, so we leave it on the granted queue and return EDEADLK in
3087	   the ast for the convert. */
3088
3089	if (deadlk && !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) {
3090		/* it's left on the granted queue */
3091		revert_lock(r, lkb);
3092		queue_cast(r, lkb, -EDEADLK);
3093		error = -EDEADLK;
3094		goto out;
3095	}
3096
3097	/* is_demoted() means the can_be_granted() above set the grmode
3098	   to NL, and left us on the granted queue.  This auto-demotion
3099	   (due to CONVDEADLK) might mean other locks, and/or this lock, are
3100	   now grantable.  We have to try to grant other converting locks
3101	   before we try again to grant this one. */
3102
3103	if (is_demoted(lkb)) {
3104		grant_pending_convert(r, DLM_LOCK_IV, NULL, NULL);
3105		if (_can_be_granted(r, lkb, 1, 0)) {
3106			grant_lock(r, lkb);
3107			queue_cast(r, lkb, 0);
3108			goto out;
3109		}
3110		/* else fall through and move to convert queue */
3111	}
3112
3113	if (can_be_queued(lkb)) {
3114		error = -EINPROGRESS;
3115		del_lkb(r, lkb);
3116		add_lkb(r, lkb, DLM_LKSTS_CONVERT);
3117		add_timeout(lkb);
3118		goto out;
3119	}
3120
3121	error = -EAGAIN;
3122	queue_cast(r, lkb, -EAGAIN);
3123 out:
3124	return error;
3125}
3126
3127static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3128			       int error)
3129{
3130	switch (error) {
3131	case 0:
3132		grant_pending_locks(r, NULL);
3133		/* grant_pending_locks also sends basts */
3134		break;
3135	case -EAGAIN:
3136		if (force_blocking_asts(lkb))
3137			send_blocking_asts_all(r, lkb);
3138		break;
3139	case -EINPROGRESS:
3140		send_blocking_asts(r, lkb);
3141		break;
3142	}
3143}
3144
3145static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3146{
3147	remove_lock(r, lkb);
3148	queue_cast(r, lkb, -DLM_EUNLOCK);
3149	return -DLM_EUNLOCK;
3150}
3151
3152static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3153			      int error)
3154{
3155	grant_pending_locks(r, NULL);
3156}
3157
3158/* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
3159
3160static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3161{
3162	int error;
3163
3164	error = revert_lock(r, lkb);
3165	if (error) {
3166		queue_cast(r, lkb, -DLM_ECANCEL);
3167		return -DLM_ECANCEL;
3168	}
3169	return 0;
3170}
3171
3172static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
3173			      int error)
3174{
3175	if (error)
3176		grant_pending_locks(r, NULL);
3177}
3178
3179/*
3180 * Four stage 3 varieties:
3181 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
3182 */
3183
3184/* add a new lkb to a possibly new rsb, called by requesting process */
3185
3186static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3187{
3188	int error;
3189
3190	/* set_master: sets lkb nodeid from r */
3191
3192	error = set_master(r, lkb);
3193	if (error < 0)
3194		goto out;
3195	if (error) {
3196		error = 0;
3197		goto out;
3198	}
3199
3200	if (is_remote(r)) {
3201		/* receive_request() calls do_request() on remote node */
3202		error = send_request(r, lkb);
3203	} else {
3204		error = do_request(r, lkb);
3205		/* for remote locks the request_reply is sent
3206		   between do_request and do_request_effects */
3207		do_request_effects(r, lkb, error);
3208	}
3209 out:
3210	return error;
3211}
3212
3213/* change some property of an existing lkb, e.g. mode */
3214
3215static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3216{
3217	int error;
3218
3219	if (is_remote(r)) {
3220		/* receive_convert() calls do_convert() on remote node */
3221		error = send_convert(r, lkb);
3222	} else {
3223		error = do_convert(r, lkb);
3224		/* for remote locks the convert_reply is sent
3225		   between do_convert and do_convert_effects */
3226		do_convert_effects(r, lkb, error);
3227	}
3228
3229	return error;
3230}
3231
3232/* remove an existing lkb from the granted queue */
3233
3234static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3235{
3236	int error;
3237
3238	if (is_remote(r)) {
3239		/* receive_unlock() calls do_unlock() on remote node */
3240		error = send_unlock(r, lkb);
3241	} else {
3242		error = do_unlock(r, lkb);
3243		/* for remote locks the unlock_reply is sent
3244		   between do_unlock and do_unlock_effects */
3245		do_unlock_effects(r, lkb, error);
3246	}
3247
3248	return error;
3249}
3250
3251/* remove an existing lkb from the convert or wait queue */
3252
3253static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3254{
3255	int error;
3256
3257	if (is_remote(r)) {
3258		/* receive_cancel() calls do_cancel() on remote node */
3259		error = send_cancel(r, lkb);
3260	} else {
3261		error = do_cancel(r, lkb);
3262		/* for remote locks the cancel_reply is sent
3263		   between do_cancel and do_cancel_effects */
3264		do_cancel_effects(r, lkb, error);
3265	}
3266
3267	return error;
3268}
3269
3270/*
3271 * Four stage 2 varieties:
3272 * request_lock(), convert_lock(), unlock_lock(), cancel_lock()
3273 */
3274
3275static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3276			const void *name, int len,
3277			struct dlm_args *args)
3278{
3279	struct dlm_rsb *r;
3280	int error;
3281
3282	error = validate_lock_args(ls, lkb, args);
3283	if (error)
3284		return error;
3285
3286	error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
3287	if (error)
3288		return error;
3289
3290	lock_rsb(r);
3291
3292	attach_lkb(r, lkb);
3293	lkb->lkb_lksb->sb_lkid = lkb->lkb_id;
3294
3295	error = _request_lock(r, lkb);
3296
3297	unlock_rsb(r);
3298	put_rsb(r);
 
 
3299	return error;
3300}
3301
3302static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3303			struct dlm_args *args)
3304{
3305	struct dlm_rsb *r;
3306	int error;
3307
3308	r = lkb->lkb_resource;
3309
3310	hold_rsb(r);
3311	lock_rsb(r);
3312
3313	error = validate_lock_args(ls, lkb, args);
3314	if (error)
3315		goto out;
3316
3317	error = _convert_lock(r, lkb);
3318 out:
3319	unlock_rsb(r);
3320	put_rsb(r);
3321	return error;
3322}
3323
3324static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3325		       struct dlm_args *args)
3326{
3327	struct dlm_rsb *r;
3328	int error;
3329
3330	r = lkb->lkb_resource;
3331
3332	hold_rsb(r);
3333	lock_rsb(r);
3334
3335	error = validate_unlock_args(lkb, args);
3336	if (error)
3337		goto out;
3338
3339	error = _unlock_lock(r, lkb);
3340 out:
3341	unlock_rsb(r);
3342	put_rsb(r);
3343	return error;
3344}
3345
3346static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb,
3347		       struct dlm_args *args)
3348{
3349	struct dlm_rsb *r;
3350	int error;
3351
3352	r = lkb->lkb_resource;
3353
3354	hold_rsb(r);
3355	lock_rsb(r);
3356
3357	error = validate_unlock_args(lkb, args);
3358	if (error)
3359		goto out;
3360
3361	error = _cancel_lock(r, lkb);
3362 out:
3363	unlock_rsb(r);
3364	put_rsb(r);
3365	return error;
3366}
3367
3368/*
3369 * Two stage 1 varieties:  dlm_lock() and dlm_unlock()
3370 */
3371
3372int dlm_lock(dlm_lockspace_t *lockspace,
3373	     int mode,
3374	     struct dlm_lksb *lksb,
3375	     uint32_t flags,
3376	     const void *name,
3377	     unsigned int namelen,
3378	     uint32_t parent_lkid,
3379	     void (*ast) (void *astarg),
3380	     void *astarg,
3381	     void (*bast) (void *astarg, int mode))
3382{
3383	struct dlm_ls *ls;
3384	struct dlm_lkb *lkb;
3385	struct dlm_args args;
3386	int error, convert = flags & DLM_LKF_CONVERT;
3387
3388	ls = dlm_find_lockspace_local(lockspace);
3389	if (!ls)
3390		return -EINVAL;
3391
3392	dlm_lock_recovery(ls);
3393
3394	if (convert)
3395		error = find_lkb(ls, lksb->sb_lkid, &lkb);
3396	else
3397		error = create_lkb(ls, &lkb);
3398
3399	if (error)
3400		goto out;
3401
3402	trace_dlm_lock_start(ls, lkb, name, namelen, mode, flags);
3403
3404#ifdef CONFIG_DLM_DEPRECATED_API
3405	error = set_lock_args(mode, lksb, flags, namelen, 0, ast,
3406			      astarg, bast, &args);
3407#else
3408	error = set_lock_args(mode, lksb, flags, namelen, ast, astarg, bast,
3409			      &args);
3410#endif
3411	if (error)
3412		goto out_put;
3413
3414	if (convert)
3415		error = convert_lock(ls, lkb, &args);
3416	else
3417		error = request_lock(ls, lkb, name, namelen, &args);
3418
3419	if (error == -EINPROGRESS)
3420		error = 0;
3421 out_put:
3422	trace_dlm_lock_end(ls, lkb, name, namelen, mode, flags, error, true);
3423
3424	if (convert || error)
3425		__put_lkb(ls, lkb);
3426	if (error == -EAGAIN || error == -EDEADLK)
3427		error = 0;
3428 out:
3429	dlm_unlock_recovery(ls);
3430	dlm_put_lockspace(ls);
3431	return error;
3432}
3433
3434int dlm_unlock(dlm_lockspace_t *lockspace,
3435	       uint32_t lkid,
3436	       uint32_t flags,
3437	       struct dlm_lksb *lksb,
3438	       void *astarg)
3439{
3440	struct dlm_ls *ls;
3441	struct dlm_lkb *lkb;
3442	struct dlm_args args;
3443	int error;
3444
3445	ls = dlm_find_lockspace_local(lockspace);
3446	if (!ls)
3447		return -EINVAL;
3448
3449	dlm_lock_recovery(ls);
3450
3451	error = find_lkb(ls, lkid, &lkb);
3452	if (error)
3453		goto out;
3454
3455	trace_dlm_unlock_start(ls, lkb, flags);
3456
3457	error = set_unlock_args(flags, astarg, &args);
3458	if (error)
3459		goto out_put;
3460
3461	if (flags & DLM_LKF_CANCEL)
3462		error = cancel_lock(ls, lkb, &args);
3463	else
3464		error = unlock_lock(ls, lkb, &args);
3465
3466	if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL)
3467		error = 0;
3468	if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)))
3469		error = 0;
3470 out_put:
3471	trace_dlm_unlock_end(ls, lkb, flags, error);
3472
3473	dlm_put_lkb(lkb);
3474 out:
3475	dlm_unlock_recovery(ls);
3476	dlm_put_lockspace(ls);
3477	return error;
3478}
3479
3480/*
3481 * send/receive routines for remote operations and replies
3482 *
3483 * send_args
3484 * send_common
3485 * send_request			receive_request
3486 * send_convert			receive_convert
3487 * send_unlock			receive_unlock
3488 * send_cancel			receive_cancel
3489 * send_grant			receive_grant
3490 * send_bast			receive_bast
3491 * send_lookup			receive_lookup
3492 * send_remove			receive_remove
3493 *
3494 * 				send_common_reply
3495 * receive_request_reply	send_request_reply
3496 * receive_convert_reply	send_convert_reply
3497 * receive_unlock_reply		send_unlock_reply
3498 * receive_cancel_reply		send_cancel_reply
3499 * receive_lookup_reply		send_lookup_reply
3500 */
3501
3502static int _create_message(struct dlm_ls *ls, int mb_len,
3503			   int to_nodeid, int mstype,
3504			   struct dlm_message **ms_ret,
3505			   struct dlm_mhandle **mh_ret,
3506			   gfp_t allocation)
3507{
3508	struct dlm_message *ms;
3509	struct dlm_mhandle *mh;
3510	char *mb;
3511
3512	/* get_buffer gives us a message handle (mh) that we need to
3513	   pass into midcomms_commit and a message buffer (mb) that we
3514	   write our data into */
3515
3516	mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, allocation, &mb);
3517	if (!mh)
3518		return -ENOBUFS;
3519
 
 
3520	ms = (struct dlm_message *) mb;
3521
3522	ms->m_header.h_version = cpu_to_le32(DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
3523	ms->m_header.u.h_lockspace = cpu_to_le32(ls->ls_global_id);
3524	ms->m_header.h_nodeid = cpu_to_le32(dlm_our_nodeid());
3525	ms->m_header.h_length = cpu_to_le16(mb_len);
3526	ms->m_header.h_cmd = DLM_MSG;
3527
3528	ms->m_type = cpu_to_le32(mstype);
3529
3530	*mh_ret = mh;
3531	*ms_ret = ms;
3532	return 0;
3533}
3534
3535static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb,
3536			  int to_nodeid, int mstype,
3537			  struct dlm_message **ms_ret,
3538			  struct dlm_mhandle **mh_ret,
3539			  gfp_t allocation)
3540{
3541	int mb_len = sizeof(struct dlm_message);
3542
3543	switch (mstype) {
3544	case DLM_MSG_REQUEST:
3545	case DLM_MSG_LOOKUP:
3546	case DLM_MSG_REMOVE:
3547		mb_len += r->res_length;
3548		break;
3549	case DLM_MSG_CONVERT:
3550	case DLM_MSG_UNLOCK:
3551	case DLM_MSG_REQUEST_REPLY:
3552	case DLM_MSG_CONVERT_REPLY:
3553	case DLM_MSG_GRANT:
3554		if (lkb && lkb->lkb_lvbptr)
3555			mb_len += r->res_ls->ls_lvblen;
3556		break;
3557	}
3558
3559	return _create_message(r->res_ls, mb_len, to_nodeid, mstype,
3560			       ms_ret, mh_ret, allocation);
3561}
3562
3563/* further lowcomms enhancements or alternate implementations may make
3564   the return value from this function useful at some point */
3565
3566static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms,
3567			const void *name, int namelen)
3568{
3569	dlm_midcomms_commit_mhandle(mh, name, namelen);
 
3570	return 0;
3571}
3572
3573static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb,
3574		      struct dlm_message *ms)
3575{
3576	ms->m_nodeid   = cpu_to_le32(lkb->lkb_nodeid);
3577	ms->m_pid      = cpu_to_le32(lkb->lkb_ownpid);
3578	ms->m_lkid     = cpu_to_le32(lkb->lkb_id);
3579	ms->m_remid    = cpu_to_le32(lkb->lkb_remid);
3580	ms->m_exflags  = cpu_to_le32(lkb->lkb_exflags);
3581	ms->m_sbflags  = cpu_to_le32(lkb->lkb_sbflags);
3582	ms->m_flags    = cpu_to_le32(lkb->lkb_flags);
3583	ms->m_lvbseq   = cpu_to_le32(lkb->lkb_lvbseq);
3584	ms->m_status   = cpu_to_le32(lkb->lkb_status);
3585	ms->m_grmode   = cpu_to_le32(lkb->lkb_grmode);
3586	ms->m_rqmode   = cpu_to_le32(lkb->lkb_rqmode);
3587	ms->m_hash     = cpu_to_le32(r->res_hash);
3588
3589	/* m_result and m_bastmode are set from function args,
3590	   not from lkb fields */
3591
3592	if (lkb->lkb_bastfn)
3593		ms->m_asts |= cpu_to_le32(DLM_CB_BAST);
3594	if (lkb->lkb_astfn)
3595		ms->m_asts |= cpu_to_le32(DLM_CB_CAST);
3596
3597	/* compare with switch in create_message; send_remove() doesn't
3598	   use send_args() */
3599
3600	switch (ms->m_type) {
3601	case cpu_to_le32(DLM_MSG_REQUEST):
3602	case cpu_to_le32(DLM_MSG_LOOKUP):
3603		memcpy(ms->m_extra, r->res_name, r->res_length);
3604		break;
3605	case cpu_to_le32(DLM_MSG_CONVERT):
3606	case cpu_to_le32(DLM_MSG_UNLOCK):
3607	case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
3608	case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
3609	case cpu_to_le32(DLM_MSG_GRANT):
3610		if (!lkb->lkb_lvbptr || !(lkb->lkb_exflags & DLM_LKF_VALBLK))
3611			break;
3612		memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen);
3613		break;
3614	}
3615}
3616
3617static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype)
3618{
3619	struct dlm_message *ms;
3620	struct dlm_mhandle *mh;
3621	int to_nodeid, error;
3622
3623	to_nodeid = r->res_nodeid;
3624
3625	error = add_to_waiters(lkb, mstype, to_nodeid);
3626	if (error)
3627		return error;
3628
3629	error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh, GFP_NOFS);
3630	if (error)
3631		goto fail;
3632
3633	send_args(r, lkb, ms);
3634
3635	error = send_message(mh, ms, r->res_name, r->res_length);
3636	if (error)
3637		goto fail;
3638	return 0;
3639
3640 fail:
3641	remove_from_waiters(lkb, msg_reply_type(mstype));
3642	return error;
3643}
3644
3645static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb)
3646{
3647	return send_common(r, lkb, DLM_MSG_REQUEST);
3648}
3649
3650static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
3651{
3652	int error;
3653
3654	error = send_common(r, lkb, DLM_MSG_CONVERT);
3655
3656	/* down conversions go without a reply from the master */
3657	if (!error && down_conversion(lkb)) {
3658		remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY);
3659		r->res_ls->ls_stub_ms.m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
3660		r->res_ls->ls_stub_ms.m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY);
3661		r->res_ls->ls_stub_ms.m_result = 0;
3662		__receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms);
3663	}
3664
3665	return error;
3666}
3667
3668/* FIXME: if this lkb is the only lock we hold on the rsb, then set
3669   MASTER_UNCERTAIN to force the next request on the rsb to confirm
3670   that the master is still correct. */
3671
3672static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
3673{
3674	return send_common(r, lkb, DLM_MSG_UNLOCK);
3675}
3676
3677static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
3678{
3679	return send_common(r, lkb, DLM_MSG_CANCEL);
3680}
3681
3682static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb)
3683{
3684	struct dlm_message *ms;
3685	struct dlm_mhandle *mh;
3686	int to_nodeid, error;
3687
3688	to_nodeid = lkb->lkb_nodeid;
3689
3690	error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh,
3691			       GFP_NOFS);
3692	if (error)
3693		goto out;
3694
3695	send_args(r, lkb, ms);
3696
3697	ms->m_result = 0;
3698
3699	error = send_message(mh, ms, r->res_name, r->res_length);
3700 out:
3701	return error;
3702}
3703
3704static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode)
3705{
3706	struct dlm_message *ms;
3707	struct dlm_mhandle *mh;
3708	int to_nodeid, error;
3709
3710	to_nodeid = lkb->lkb_nodeid;
3711
3712	error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh,
3713			       GFP_NOFS);
3714	if (error)
3715		goto out;
3716
3717	send_args(r, lkb, ms);
3718
3719	ms->m_bastmode = cpu_to_le32(mode);
3720
3721	error = send_message(mh, ms, r->res_name, r->res_length);
3722 out:
3723	return error;
3724}
3725
3726static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb)
3727{
3728	struct dlm_message *ms;
3729	struct dlm_mhandle *mh;
3730	int to_nodeid, error;
3731
3732	to_nodeid = dlm_dir_nodeid(r);
3733
3734	error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid);
3735	if (error)
3736		return error;
3737
3738	error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh,
3739			       GFP_NOFS);
3740	if (error)
3741		goto fail;
3742
3743	send_args(r, lkb, ms);
3744
3745	error = send_message(mh, ms, r->res_name, r->res_length);
3746	if (error)
3747		goto fail;
3748	return 0;
3749
3750 fail:
3751	remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
3752	return error;
3753}
3754
3755static int send_remove(struct dlm_rsb *r)
3756{
3757	struct dlm_message *ms;
3758	struct dlm_mhandle *mh;
3759	int to_nodeid, error;
3760
3761	to_nodeid = dlm_dir_nodeid(r);
3762
3763	error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh,
3764			       GFP_ATOMIC);
3765	if (error)
3766		goto out;
3767
3768	memcpy(ms->m_extra, r->res_name, r->res_length);
3769	ms->m_hash = cpu_to_le32(r->res_hash);
3770
3771	error = send_message(mh, ms, r->res_name, r->res_length);
3772 out:
3773	return error;
3774}
3775
3776static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
3777			     int mstype, int rv)
3778{
3779	struct dlm_message *ms;
3780	struct dlm_mhandle *mh;
3781	int to_nodeid, error;
3782
3783	to_nodeid = lkb->lkb_nodeid;
3784
3785	error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh, GFP_NOFS);
3786	if (error)
3787		goto out;
3788
3789	send_args(r, lkb, ms);
3790
3791	ms->m_result = cpu_to_le32(to_dlm_errno(rv));
3792
3793	error = send_message(mh, ms, r->res_name, r->res_length);
3794 out:
3795	return error;
3796}
3797
3798static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3799{
3800	return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv);
3801}
3802
3803static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3804{
3805	return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv);
3806}
3807
3808static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3809{
3810	return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv);
3811}
3812
3813static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv)
3814{
3815	return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv);
3816}
3817
3818static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in,
3819			     int ret_nodeid, int rv)
3820{
3821	struct dlm_rsb *r = &ls->ls_stub_rsb;
3822	struct dlm_message *ms;
3823	struct dlm_mhandle *mh;
3824	int error, nodeid = le32_to_cpu(ms_in->m_header.h_nodeid);
3825
3826	error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh,
3827			       GFP_NOFS);
3828	if (error)
3829		goto out;
3830
3831	ms->m_lkid = ms_in->m_lkid;
3832	ms->m_result = cpu_to_le32(to_dlm_errno(rv));
3833	ms->m_nodeid = cpu_to_le32(ret_nodeid);
3834
3835	error = send_message(mh, ms, ms_in->m_extra, receive_extralen(ms_in));
3836 out:
3837	return error;
3838}
3839
3840/* which args we save from a received message depends heavily on the type
3841   of message, unlike the send side where we can safely send everything about
3842   the lkb for any type of message */
3843
3844static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms)
3845{
3846	lkb->lkb_exflags = le32_to_cpu(ms->m_exflags);
3847	lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags);
3848	lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3849			  (le32_to_cpu(ms->m_flags) & 0x0000FFFF);
3850}
3851
3852static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
3853{
3854	if (ms->m_flags == cpu_to_le32(DLM_IFL_STUB_MS))
3855		return;
3856
3857	lkb->lkb_sbflags = le32_to_cpu(ms->m_sbflags);
3858	lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) |
3859			 (le32_to_cpu(ms->m_flags) & 0x0000FFFF);
3860}
3861
3862static int receive_extralen(struct dlm_message *ms)
3863{
3864	return (le16_to_cpu(ms->m_header.h_length) -
3865		sizeof(struct dlm_message));
3866}
3867
3868static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb,
3869		       struct dlm_message *ms)
3870{
3871	int len;
3872
3873	if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3874		if (!lkb->lkb_lvbptr)
3875			lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3876		if (!lkb->lkb_lvbptr)
3877			return -ENOMEM;
3878		len = receive_extralen(ms);
3879		if (len > ls->ls_lvblen)
3880			len = ls->ls_lvblen;
3881		memcpy(lkb->lkb_lvbptr, ms->m_extra, len);
3882	}
3883	return 0;
3884}
3885
3886static void fake_bastfn(void *astparam, int mode)
3887{
3888	log_print("fake_bastfn should not be called");
3889}
3890
3891static void fake_astfn(void *astparam)
3892{
3893	log_print("fake_astfn should not be called");
3894}
3895
3896static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3897				struct dlm_message *ms)
3898{
3899	lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
3900	lkb->lkb_ownpid = le32_to_cpu(ms->m_pid);
3901	lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
3902	lkb->lkb_grmode = DLM_LOCK_IV;
3903	lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
3904
3905	lkb->lkb_bastfn = (ms->m_asts & cpu_to_le32(DLM_CB_BAST)) ? &fake_bastfn : NULL;
3906	lkb->lkb_astfn = (ms->m_asts & cpu_to_le32(DLM_CB_CAST)) ? &fake_astfn : NULL;
3907
3908	if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
3909		/* lkb was just created so there won't be an lvb yet */
3910		lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
3911		if (!lkb->lkb_lvbptr)
3912			return -ENOMEM;
3913	}
3914
3915	return 0;
3916}
3917
3918static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3919				struct dlm_message *ms)
3920{
3921	if (lkb->lkb_status != DLM_LKSTS_GRANTED)
3922		return -EBUSY;
3923
3924	if (receive_lvb(ls, lkb, ms))
3925		return -ENOMEM;
3926
3927	lkb->lkb_rqmode = le32_to_cpu(ms->m_rqmode);
3928	lkb->lkb_lvbseq = le32_to_cpu(ms->m_lvbseq);
3929
3930	return 0;
3931}
3932
3933static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
3934			       struct dlm_message *ms)
3935{
3936	if (receive_lvb(ls, lkb, ms))
3937		return -ENOMEM;
3938	return 0;
3939}
3940
3941/* We fill in the stub-lkb fields with the info that send_xxxx_reply()
3942   uses to send a reply and that the remote end uses to process the reply. */
3943
3944static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms)
3945{
3946	struct dlm_lkb *lkb = &ls->ls_stub_lkb;
3947	lkb->lkb_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
3948	lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
3949}
3950
3951/* This is called after the rsb is locked so that we can safely inspect
3952   fields in the lkb. */
3953
3954static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms)
3955{
3956	int from = le32_to_cpu(ms->m_header.h_nodeid);
3957	int error = 0;
3958
3959	/* currently mixing of user/kernel locks are not supported */
3960	if (ms->m_flags & cpu_to_le32(DLM_IFL_USER) &&
3961	    ~lkb->lkb_flags & DLM_IFL_USER) {
3962		log_error(lkb->lkb_resource->res_ls,
3963			  "got user dlm message for a kernel lock");
3964		error = -EINVAL;
3965		goto out;
3966	}
3967
3968	switch (ms->m_type) {
3969	case cpu_to_le32(DLM_MSG_CONVERT):
3970	case cpu_to_le32(DLM_MSG_UNLOCK):
3971	case cpu_to_le32(DLM_MSG_CANCEL):
3972		if (!is_master_copy(lkb) || lkb->lkb_nodeid != from)
3973			error = -EINVAL;
3974		break;
3975
3976	case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
3977	case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
3978	case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
3979	case cpu_to_le32(DLM_MSG_GRANT):
3980	case cpu_to_le32(DLM_MSG_BAST):
3981		if (!is_process_copy(lkb) || lkb->lkb_nodeid != from)
3982			error = -EINVAL;
3983		break;
3984
3985	case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
3986		if (!is_process_copy(lkb))
3987			error = -EINVAL;
3988		else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from)
3989			error = -EINVAL;
3990		break;
3991
3992	default:
3993		error = -EINVAL;
3994	}
3995
3996out:
3997	if (error)
3998		log_error(lkb->lkb_resource->res_ls,
3999			  "ignore invalid message %d from %d %x %x %x %d",
4000			  le32_to_cpu(ms->m_type), from, lkb->lkb_id,
4001			  lkb->lkb_remid, lkb->lkb_flags, lkb->lkb_nodeid);
4002	return error;
4003}
4004
4005static int receive_request(struct dlm_ls *ls, struct dlm_message *ms)
4006{
4007	struct dlm_lkb *lkb;
4008	struct dlm_rsb *r;
4009	int from_nodeid;
4010	int error, namelen = 0;
4011
4012	from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4013
4014	error = create_lkb(ls, &lkb);
4015	if (error)
4016		goto fail;
4017
4018	receive_flags(lkb, ms);
4019	lkb->lkb_flags |= DLM_IFL_MSTCPY;
4020	error = receive_request_args(ls, lkb, ms);
4021	if (error) {
4022		__put_lkb(ls, lkb);
4023		goto fail;
4024	}
4025
4026	/* The dir node is the authority on whether we are the master
4027	   for this rsb or not, so if the master sends us a request, we should
4028	   recreate the rsb if we've destroyed it.   This race happens when we
4029	   send a remove message to the dir node at the same time that the dir
4030	   node sends us a request for the rsb. */
4031
4032	namelen = receive_extralen(ms);
4033
4034	error = find_rsb(ls, ms->m_extra, namelen, from_nodeid,
4035			 R_RECEIVE_REQUEST, &r);
4036	if (error) {
4037		__put_lkb(ls, lkb);
4038		goto fail;
4039	}
4040
4041	lock_rsb(r);
4042
4043	if (r->res_master_nodeid != dlm_our_nodeid()) {
4044		error = validate_master_nodeid(ls, r, from_nodeid);
4045		if (error) {
4046			unlock_rsb(r);
4047			put_rsb(r);
4048			__put_lkb(ls, lkb);
4049			goto fail;
4050		}
4051	}
4052
4053	attach_lkb(r, lkb);
4054	error = do_request(r, lkb);
4055	send_request_reply(r, lkb, error);
4056	do_request_effects(r, lkb, error);
4057
4058	unlock_rsb(r);
4059	put_rsb(r);
4060
4061	if (error == -EINPROGRESS)
4062		error = 0;
4063	if (error)
4064		dlm_put_lkb(lkb);
4065	return 0;
4066
4067 fail:
4068	/* TODO: instead of returning ENOTBLK, add the lkb to res_lookup
4069	   and do this receive_request again from process_lookup_list once
4070	   we get the lookup reply.  This would avoid a many repeated
4071	   ENOTBLK request failures when the lookup reply designating us
4072	   as master is delayed. */
4073
4074	if (error != -ENOTBLK) {
4075		log_limit(ls, "receive_request %x from %d %d",
4076			  le32_to_cpu(ms->m_lkid), from_nodeid, error);
4077	}
4078
4079	setup_stub_lkb(ls, ms);
4080	send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4081	return error;
4082}
4083
4084static int receive_convert(struct dlm_ls *ls, struct dlm_message *ms)
4085{
4086	struct dlm_lkb *lkb;
4087	struct dlm_rsb *r;
4088	int error, reply = 1;
4089
4090	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4091	if (error)
4092		goto fail;
4093
4094	if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
4095		log_error(ls, "receive_convert %x remid %x recover_seq %llu "
4096			  "remote %d %x", lkb->lkb_id, lkb->lkb_remid,
4097			  (unsigned long long)lkb->lkb_recover_seq,
4098			  le32_to_cpu(ms->m_header.h_nodeid),
4099			  le32_to_cpu(ms->m_lkid));
4100		error = -ENOENT;
4101		dlm_put_lkb(lkb);
4102		goto fail;
4103	}
4104
4105	r = lkb->lkb_resource;
4106
4107	hold_rsb(r);
4108	lock_rsb(r);
4109
4110	error = validate_message(lkb, ms);
4111	if (error)
4112		goto out;
4113
4114	receive_flags(lkb, ms);
4115
4116	error = receive_convert_args(ls, lkb, ms);
4117	if (error) {
4118		send_convert_reply(r, lkb, error);
4119		goto out;
4120	}
4121
4122	reply = !down_conversion(lkb);
4123
4124	error = do_convert(r, lkb);
4125	if (reply)
4126		send_convert_reply(r, lkb, error);
4127	do_convert_effects(r, lkb, error);
4128 out:
4129	unlock_rsb(r);
4130	put_rsb(r);
4131	dlm_put_lkb(lkb);
4132	return 0;
4133
4134 fail:
4135	setup_stub_lkb(ls, ms);
4136	send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4137	return error;
4138}
4139
4140static int receive_unlock(struct dlm_ls *ls, struct dlm_message *ms)
4141{
4142	struct dlm_lkb *lkb;
4143	struct dlm_rsb *r;
4144	int error;
4145
4146	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4147	if (error)
4148		goto fail;
4149
4150	if (lkb->lkb_remid != le32_to_cpu(ms->m_lkid)) {
4151		log_error(ls, "receive_unlock %x remid %x remote %d %x",
4152			  lkb->lkb_id, lkb->lkb_remid,
4153			  le32_to_cpu(ms->m_header.h_nodeid),
4154			  le32_to_cpu(ms->m_lkid));
4155		error = -ENOENT;
4156		dlm_put_lkb(lkb);
4157		goto fail;
4158	}
4159
4160	r = lkb->lkb_resource;
4161
4162	hold_rsb(r);
4163	lock_rsb(r);
4164
4165	error = validate_message(lkb, ms);
4166	if (error)
4167		goto out;
4168
4169	receive_flags(lkb, ms);
4170
4171	error = receive_unlock_args(ls, lkb, ms);
4172	if (error) {
4173		send_unlock_reply(r, lkb, error);
4174		goto out;
4175	}
4176
4177	error = do_unlock(r, lkb);
4178	send_unlock_reply(r, lkb, error);
4179	do_unlock_effects(r, lkb, error);
4180 out:
4181	unlock_rsb(r);
4182	put_rsb(r);
4183	dlm_put_lkb(lkb);
4184	return 0;
4185
4186 fail:
4187	setup_stub_lkb(ls, ms);
4188	send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4189	return error;
4190}
4191
4192static int receive_cancel(struct dlm_ls *ls, struct dlm_message *ms)
4193{
4194	struct dlm_lkb *lkb;
4195	struct dlm_rsb *r;
4196	int error;
4197
4198	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4199	if (error)
4200		goto fail;
4201
4202	receive_flags(lkb, ms);
4203
4204	r = lkb->lkb_resource;
4205
4206	hold_rsb(r);
4207	lock_rsb(r);
4208
4209	error = validate_message(lkb, ms);
4210	if (error)
4211		goto out;
4212
4213	error = do_cancel(r, lkb);
4214	send_cancel_reply(r, lkb, error);
4215	do_cancel_effects(r, lkb, error);
4216 out:
4217	unlock_rsb(r);
4218	put_rsb(r);
4219	dlm_put_lkb(lkb);
4220	return 0;
4221
4222 fail:
4223	setup_stub_lkb(ls, ms);
4224	send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error);
4225	return error;
4226}
4227
4228static int receive_grant(struct dlm_ls *ls, struct dlm_message *ms)
4229{
4230	struct dlm_lkb *lkb;
4231	struct dlm_rsb *r;
4232	int error;
4233
4234	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4235	if (error)
4236		return error;
4237
4238	r = lkb->lkb_resource;
4239
4240	hold_rsb(r);
4241	lock_rsb(r);
4242
4243	error = validate_message(lkb, ms);
4244	if (error)
4245		goto out;
4246
4247	receive_flags_reply(lkb, ms);
4248	if (is_altmode(lkb))
4249		munge_altmode(lkb, ms);
4250	grant_lock_pc(r, lkb, ms);
4251	queue_cast(r, lkb, 0);
4252 out:
4253	unlock_rsb(r);
4254	put_rsb(r);
4255	dlm_put_lkb(lkb);
4256	return 0;
4257}
4258
4259static int receive_bast(struct dlm_ls *ls, struct dlm_message *ms)
4260{
4261	struct dlm_lkb *lkb;
4262	struct dlm_rsb *r;
4263	int error;
4264
4265	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4266	if (error)
4267		return error;
4268
4269	r = lkb->lkb_resource;
4270
4271	hold_rsb(r);
4272	lock_rsb(r);
4273
4274	error = validate_message(lkb, ms);
4275	if (error)
4276		goto out;
4277
4278	queue_bast(r, lkb, le32_to_cpu(ms->m_bastmode));
4279	lkb->lkb_highbast = le32_to_cpu(ms->m_bastmode);
4280 out:
4281	unlock_rsb(r);
4282	put_rsb(r);
4283	dlm_put_lkb(lkb);
4284	return 0;
4285}
4286
4287static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms)
4288{
4289	int len, error, ret_nodeid, from_nodeid, our_nodeid;
4290
4291	from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4292	our_nodeid = dlm_our_nodeid();
4293
4294	len = receive_extralen(ms);
4295
4296	error = dlm_master_lookup(ls, from_nodeid, ms->m_extra, len, 0,
4297				  &ret_nodeid, NULL);
 
 
 
 
 
 
 
 
4298
4299	/* Optimization: we're master so treat lookup as a request */
4300	if (!error && ret_nodeid == our_nodeid) {
4301		receive_request(ls, ms);
4302		return;
4303	}
 
4304	send_lookup_reply(ls, ms, ret_nodeid, error);
4305}
4306
4307static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms)
4308{
4309	char name[DLM_RESNAME_MAXLEN+1];
4310	struct dlm_rsb *r;
4311	uint32_t hash, b;
4312	int rv, len, dir_nodeid, from_nodeid;
4313
4314	from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4315
4316	len = receive_extralen(ms);
4317
4318	if (len > DLM_RESNAME_MAXLEN) {
4319		log_error(ls, "receive_remove from %d bad len %d",
4320			  from_nodeid, len);
4321		return;
4322	}
4323
4324	dir_nodeid = dlm_hash2nodeid(ls, le32_to_cpu(ms->m_hash));
4325	if (dir_nodeid != dlm_our_nodeid()) {
4326		log_error(ls, "receive_remove from %d bad nodeid %d",
4327			  from_nodeid, dir_nodeid);
4328		return;
4329	}
4330
4331	/* Look for name on rsbtbl.toss, if it's there, kill it.
4332	   If it's on rsbtbl.keep, it's being used, and we should ignore this
4333	   message.  This is an expected race between the dir node sending a
4334	   request to the master node at the same time as the master node sends
4335	   a remove to the dir node.  The resolution to that race is for the
4336	   dir node to ignore the remove message, and the master node to
4337	   recreate the master rsb when it gets a request from the dir node for
4338	   an rsb it doesn't have. */
4339
4340	memset(name, 0, sizeof(name));
4341	memcpy(name, ms->m_extra, len);
4342
4343	hash = jhash(name, len, 0);
4344	b = hash & (ls->ls_rsbtbl_size - 1);
4345
4346	spin_lock(&ls->ls_rsbtbl[b].lock);
4347
4348	rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].toss, name, len, &r);
4349	if (rv) {
4350		/* verify the rsb is on keep list per comment above */
4351		rv = dlm_search_rsb_tree(&ls->ls_rsbtbl[b].keep, name, len, &r);
4352		if (rv) {
4353			/* should not happen */
4354			log_error(ls, "receive_remove from %d not found %s",
4355				  from_nodeid, name);
4356			spin_unlock(&ls->ls_rsbtbl[b].lock);
4357			return;
4358		}
4359		if (r->res_master_nodeid != from_nodeid) {
4360			/* should not happen */
4361			log_error(ls, "receive_remove keep from %d master %d",
4362				  from_nodeid, r->res_master_nodeid);
4363			dlm_print_rsb(r);
4364			spin_unlock(&ls->ls_rsbtbl[b].lock);
4365			return;
4366		}
4367
4368		log_debug(ls, "receive_remove from %d master %d first %x %s",
4369			  from_nodeid, r->res_master_nodeid, r->res_first_lkid,
4370			  name);
4371		spin_unlock(&ls->ls_rsbtbl[b].lock);
4372		return;
4373	}
4374
4375	if (r->res_master_nodeid != from_nodeid) {
4376		log_error(ls, "receive_remove toss from %d master %d",
4377			  from_nodeid, r->res_master_nodeid);
4378		dlm_print_rsb(r);
4379		spin_unlock(&ls->ls_rsbtbl[b].lock);
4380		return;
4381	}
4382
4383	if (kref_put(&r->res_ref, kill_rsb)) {
4384		rb_erase(&r->res_hashnode, &ls->ls_rsbtbl[b].toss);
4385		spin_unlock(&ls->ls_rsbtbl[b].lock);
4386		dlm_free_rsb(r);
4387	} else {
4388		log_error(ls, "receive_remove from %d rsb ref error",
4389			  from_nodeid);
4390		dlm_print_rsb(r);
4391		spin_unlock(&ls->ls_rsbtbl[b].lock);
4392	}
4393}
4394
4395static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms)
4396{
4397	do_purge(ls, le32_to_cpu(ms->m_nodeid), le32_to_cpu(ms->m_pid));
4398}
4399
4400static int receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms)
4401{
4402	struct dlm_lkb *lkb;
4403	struct dlm_rsb *r;
4404	int error, mstype, result;
4405	int from_nodeid = le32_to_cpu(ms->m_header.h_nodeid);
4406
4407	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4408	if (error)
4409		return error;
4410
4411	r = lkb->lkb_resource;
4412	hold_rsb(r);
4413	lock_rsb(r);
4414
4415	error = validate_message(lkb, ms);
4416	if (error)
4417		goto out;
4418
4419	mstype = lkb->lkb_wait_type;
4420	error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY);
4421	if (error) {
4422		log_error(ls, "receive_request_reply %x remote %d %x result %d",
4423			  lkb->lkb_id, from_nodeid, le32_to_cpu(ms->m_lkid),
4424			  from_dlm_errno(le32_to_cpu(ms->m_result)));
4425		dlm_dump_rsb(r);
4426		goto out;
4427	}
4428
4429	/* Optimization: the dir node was also the master, so it took our
4430	   lookup as a request and sent request reply instead of lookup reply */
4431	if (mstype == DLM_MSG_LOOKUP) {
4432		r->res_master_nodeid = from_nodeid;
4433		r->res_nodeid = from_nodeid;
4434		lkb->lkb_nodeid = from_nodeid;
4435	}
4436
4437	/* this is the value returned from do_request() on the master */
4438	result = from_dlm_errno(le32_to_cpu(ms->m_result));
4439
4440	switch (result) {
4441	case -EAGAIN:
4442		/* request would block (be queued) on remote master */
4443		queue_cast(r, lkb, -EAGAIN);
4444		confirm_master(r, -EAGAIN);
4445		unhold_lkb(lkb); /* undoes create_lkb() */
4446		break;
4447
4448	case -EINPROGRESS:
4449	case 0:
4450		/* request was queued or granted on remote master */
4451		receive_flags_reply(lkb, ms);
4452		lkb->lkb_remid = le32_to_cpu(ms->m_lkid);
4453		if (is_altmode(lkb))
4454			munge_altmode(lkb, ms);
4455		if (result) {
4456			add_lkb(r, lkb, DLM_LKSTS_WAITING);
4457			add_timeout(lkb);
4458		} else {
4459			grant_lock_pc(r, lkb, ms);
4460			queue_cast(r, lkb, 0);
4461		}
4462		confirm_master(r, result);
4463		break;
4464
4465	case -EBADR:
4466	case -ENOTBLK:
4467		/* find_rsb failed to find rsb or rsb wasn't master */
4468		log_limit(ls, "receive_request_reply %x from %d %d "
4469			  "master %d dir %d first %x %s", lkb->lkb_id,
4470			  from_nodeid, result, r->res_master_nodeid,
4471			  r->res_dir_nodeid, r->res_first_lkid, r->res_name);
4472
4473		if (r->res_dir_nodeid != dlm_our_nodeid() &&
4474		    r->res_master_nodeid != dlm_our_nodeid()) {
4475			/* cause _request_lock->set_master->send_lookup */
4476			r->res_master_nodeid = 0;
4477			r->res_nodeid = -1;
4478			lkb->lkb_nodeid = -1;
4479		}
4480
4481		if (is_overlap(lkb)) {
4482			/* we'll ignore error in cancel/unlock reply */
4483			queue_cast_overlap(r, lkb);
4484			confirm_master(r, result);
4485			unhold_lkb(lkb); /* undoes create_lkb() */
4486		} else {
4487			_request_lock(r, lkb);
4488
4489			if (r->res_master_nodeid == dlm_our_nodeid())
4490				confirm_master(r, 0);
4491		}
4492		break;
4493
4494	default:
4495		log_error(ls, "receive_request_reply %x error %d",
4496			  lkb->lkb_id, result);
4497	}
4498
4499	if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) {
4500		log_debug(ls, "receive_request_reply %x result %d unlock",
4501			  lkb->lkb_id, result);
4502		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4503		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4504		send_unlock(r, lkb);
4505	} else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) {
4506		log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id);
4507		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4508		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4509		send_cancel(r, lkb);
4510	} else {
4511		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
4512		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
4513	}
4514 out:
4515	unlock_rsb(r);
4516	put_rsb(r);
4517	dlm_put_lkb(lkb);
4518	return 0;
4519}
4520
4521static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb,
4522				    struct dlm_message *ms)
4523{
4524	/* this is the value returned from do_convert() on the master */
4525	switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4526	case -EAGAIN:
4527		/* convert would block (be queued) on remote master */
4528		queue_cast(r, lkb, -EAGAIN);
4529		break;
4530
4531	case -EDEADLK:
4532		receive_flags_reply(lkb, ms);
4533		revert_lock_pc(r, lkb);
4534		queue_cast(r, lkb, -EDEADLK);
4535		break;
4536
4537	case -EINPROGRESS:
4538		/* convert was queued on remote master */
4539		receive_flags_reply(lkb, ms);
4540		if (is_demoted(lkb))
4541			munge_demoted(lkb);
4542		del_lkb(r, lkb);
4543		add_lkb(r, lkb, DLM_LKSTS_CONVERT);
4544		add_timeout(lkb);
4545		break;
4546
4547	case 0:
4548		/* convert was granted on remote master */
4549		receive_flags_reply(lkb, ms);
4550		if (is_demoted(lkb))
4551			munge_demoted(lkb);
4552		grant_lock_pc(r, lkb, ms);
4553		queue_cast(r, lkb, 0);
4554		break;
4555
4556	default:
4557		log_error(r->res_ls, "receive_convert_reply %x remote %d %x %d",
4558			  lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
4559			  le32_to_cpu(ms->m_lkid),
4560			  from_dlm_errno(le32_to_cpu(ms->m_result)));
4561		dlm_print_rsb(r);
4562		dlm_print_lkb(lkb);
4563	}
4564}
4565
4566static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4567{
4568	struct dlm_rsb *r = lkb->lkb_resource;
4569	int error;
4570
4571	hold_rsb(r);
4572	lock_rsb(r);
4573
4574	error = validate_message(lkb, ms);
4575	if (error)
4576		goto out;
4577
4578	/* stub reply can happen with waiters_mutex held */
4579	error = remove_from_waiters_ms(lkb, ms);
4580	if (error)
4581		goto out;
4582
4583	__receive_convert_reply(r, lkb, ms);
4584 out:
4585	unlock_rsb(r);
4586	put_rsb(r);
4587}
4588
4589static int receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms)
4590{
4591	struct dlm_lkb *lkb;
4592	int error;
4593
4594	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4595	if (error)
4596		return error;
4597
4598	_receive_convert_reply(lkb, ms);
4599	dlm_put_lkb(lkb);
4600	return 0;
4601}
4602
4603static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4604{
4605	struct dlm_rsb *r = lkb->lkb_resource;
4606	int error;
4607
4608	hold_rsb(r);
4609	lock_rsb(r);
4610
4611	error = validate_message(lkb, ms);
4612	if (error)
4613		goto out;
4614
4615	/* stub reply can happen with waiters_mutex held */
4616	error = remove_from_waiters_ms(lkb, ms);
4617	if (error)
4618		goto out;
4619
4620	/* this is the value returned from do_unlock() on the master */
4621
4622	switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4623	case -DLM_EUNLOCK:
4624		receive_flags_reply(lkb, ms);
4625		remove_lock_pc(r, lkb);
4626		queue_cast(r, lkb, -DLM_EUNLOCK);
4627		break;
4628	case -ENOENT:
4629		break;
4630	default:
4631		log_error(r->res_ls, "receive_unlock_reply %x error %d",
4632			  lkb->lkb_id, from_dlm_errno(le32_to_cpu(ms->m_result)));
4633	}
4634 out:
4635	unlock_rsb(r);
4636	put_rsb(r);
4637}
4638
4639static int receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms)
4640{
4641	struct dlm_lkb *lkb;
4642	int error;
4643
4644	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4645	if (error)
4646		return error;
4647
4648	_receive_unlock_reply(lkb, ms);
4649	dlm_put_lkb(lkb);
4650	return 0;
4651}
4652
4653static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms)
4654{
4655	struct dlm_rsb *r = lkb->lkb_resource;
4656	int error;
4657
4658	hold_rsb(r);
4659	lock_rsb(r);
4660
4661	error = validate_message(lkb, ms);
4662	if (error)
4663		goto out;
4664
4665	/* stub reply can happen with waiters_mutex held */
4666	error = remove_from_waiters_ms(lkb, ms);
4667	if (error)
4668		goto out;
4669
4670	/* this is the value returned from do_cancel() on the master */
4671
4672	switch (from_dlm_errno(le32_to_cpu(ms->m_result))) {
4673	case -DLM_ECANCEL:
4674		receive_flags_reply(lkb, ms);
4675		revert_lock_pc(r, lkb);
4676		queue_cast(r, lkb, -DLM_ECANCEL);
4677		break;
4678	case 0:
4679		break;
4680	default:
4681		log_error(r->res_ls, "receive_cancel_reply %x error %d",
4682			  lkb->lkb_id,
4683			  from_dlm_errno(le32_to_cpu(ms->m_result)));
4684	}
4685 out:
4686	unlock_rsb(r);
4687	put_rsb(r);
4688}
4689
4690static int receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms)
4691{
4692	struct dlm_lkb *lkb;
4693	int error;
4694
4695	error = find_lkb(ls, le32_to_cpu(ms->m_remid), &lkb);
4696	if (error)
4697		return error;
4698
4699	_receive_cancel_reply(lkb, ms);
4700	dlm_put_lkb(lkb);
4701	return 0;
4702}
4703
4704static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms)
4705{
4706	struct dlm_lkb *lkb;
4707	struct dlm_rsb *r;
4708	int error, ret_nodeid;
4709	int do_lookup_list = 0;
4710
4711	error = find_lkb(ls, le32_to_cpu(ms->m_lkid), &lkb);
4712	if (error) {
4713		log_error(ls, "%s no lkid %x", __func__,
4714			  le32_to_cpu(ms->m_lkid));
4715		return;
4716	}
4717
4718	/* ms->m_result is the value returned by dlm_master_lookup on dir node
4719	   FIXME: will a non-zero error ever be returned? */
4720
4721	r = lkb->lkb_resource;
4722	hold_rsb(r);
4723	lock_rsb(r);
4724
4725	error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY);
4726	if (error)
4727		goto out;
4728
4729	ret_nodeid = le32_to_cpu(ms->m_nodeid);
4730
4731	/* We sometimes receive a request from the dir node for this
4732	   rsb before we've received the dir node's loookup_reply for it.
4733	   The request from the dir node implies we're the master, so we set
4734	   ourself as master in receive_request_reply, and verify here that
4735	   we are indeed the master. */
4736
4737	if (r->res_master_nodeid && (r->res_master_nodeid != ret_nodeid)) {
4738		/* This should never happen */
4739		log_error(ls, "receive_lookup_reply %x from %d ret %d "
4740			  "master %d dir %d our %d first %x %s",
4741			  lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid),
4742			  ret_nodeid, r->res_master_nodeid, r->res_dir_nodeid,
4743			  dlm_our_nodeid(), r->res_first_lkid, r->res_name);
4744	}
4745
4746	if (ret_nodeid == dlm_our_nodeid()) {
4747		r->res_master_nodeid = ret_nodeid;
4748		r->res_nodeid = 0;
4749		do_lookup_list = 1;
4750		r->res_first_lkid = 0;
4751	} else if (ret_nodeid == -1) {
4752		/* the remote node doesn't believe it's the dir node */
4753		log_error(ls, "receive_lookup_reply %x from %d bad ret_nodeid",
4754			  lkb->lkb_id, le32_to_cpu(ms->m_header.h_nodeid));
4755		r->res_master_nodeid = 0;
4756		r->res_nodeid = -1;
4757		lkb->lkb_nodeid = -1;
4758	} else {
4759		/* set_master() will set lkb_nodeid from r */
4760		r->res_master_nodeid = ret_nodeid;
4761		r->res_nodeid = ret_nodeid;
4762	}
4763
4764	if (is_overlap(lkb)) {
4765		log_debug(ls, "receive_lookup_reply %x unlock %x",
4766			  lkb->lkb_id, lkb->lkb_flags);
4767		queue_cast_overlap(r, lkb);
4768		unhold_lkb(lkb); /* undoes create_lkb() */
4769		goto out_list;
4770	}
4771
4772	_request_lock(r, lkb);
4773
4774 out_list:
4775	if (do_lookup_list)
4776		process_lookup_list(r);
4777 out:
4778	unlock_rsb(r);
4779	put_rsb(r);
4780	dlm_put_lkb(lkb);
4781}
4782
4783static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4784			     uint32_t saved_seq)
4785{
4786	int error = 0, noent = 0;
4787
4788	if (!dlm_is_member(ls, le32_to_cpu(ms->m_header.h_nodeid))) {
4789		log_limit(ls, "receive %d from non-member %d %x %x %d",
4790			  le32_to_cpu(ms->m_type),
4791			  le32_to_cpu(ms->m_header.h_nodeid),
4792			  le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
4793			  from_dlm_errno(le32_to_cpu(ms->m_result)));
4794		return;
4795	}
4796
4797	switch (ms->m_type) {
4798
4799	/* messages sent to a master node */
4800
4801	case cpu_to_le32(DLM_MSG_REQUEST):
4802		error = receive_request(ls, ms);
4803		break;
4804
4805	case cpu_to_le32(DLM_MSG_CONVERT):
4806		error = receive_convert(ls, ms);
4807		break;
4808
4809	case cpu_to_le32(DLM_MSG_UNLOCK):
4810		error = receive_unlock(ls, ms);
4811		break;
4812
4813	case cpu_to_le32(DLM_MSG_CANCEL):
4814		noent = 1;
4815		error = receive_cancel(ls, ms);
4816		break;
4817
4818	/* messages sent from a master node (replies to above) */
4819
4820	case cpu_to_le32(DLM_MSG_REQUEST_REPLY):
4821		error = receive_request_reply(ls, ms);
4822		break;
4823
4824	case cpu_to_le32(DLM_MSG_CONVERT_REPLY):
4825		error = receive_convert_reply(ls, ms);
4826		break;
4827
4828	case cpu_to_le32(DLM_MSG_UNLOCK_REPLY):
4829		error = receive_unlock_reply(ls, ms);
4830		break;
4831
4832	case cpu_to_le32(DLM_MSG_CANCEL_REPLY):
4833		error = receive_cancel_reply(ls, ms);
4834		break;
4835
4836	/* messages sent from a master node (only two types of async msg) */
4837
4838	case cpu_to_le32(DLM_MSG_GRANT):
4839		noent = 1;
4840		error = receive_grant(ls, ms);
4841		break;
4842
4843	case cpu_to_le32(DLM_MSG_BAST):
4844		noent = 1;
4845		error = receive_bast(ls, ms);
4846		break;
4847
4848	/* messages sent to a dir node */
4849
4850	case cpu_to_le32(DLM_MSG_LOOKUP):
4851		receive_lookup(ls, ms);
4852		break;
4853
4854	case cpu_to_le32(DLM_MSG_REMOVE):
4855		receive_remove(ls, ms);
4856		break;
4857
4858	/* messages sent from a dir node (remove has no reply) */
4859
4860	case cpu_to_le32(DLM_MSG_LOOKUP_REPLY):
4861		receive_lookup_reply(ls, ms);
4862		break;
4863
4864	/* other messages */
4865
4866	case cpu_to_le32(DLM_MSG_PURGE):
4867		receive_purge(ls, ms);
4868		break;
4869
4870	default:
4871		log_error(ls, "unknown message type %d",
4872			  le32_to_cpu(ms->m_type));
4873	}
4874
4875	/*
4876	 * When checking for ENOENT, we're checking the result of
4877	 * find_lkb(m_remid):
4878	 *
4879	 * The lock id referenced in the message wasn't found.  This may
4880	 * happen in normal usage for the async messages and cancel, so
4881	 * only use log_debug for them.
4882	 *
4883	 * Some errors are expected and normal.
4884	 */
4885
4886	if (error == -ENOENT && noent) {
4887		log_debug(ls, "receive %d no %x remote %d %x saved_seq %u",
4888			  le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
4889			  le32_to_cpu(ms->m_header.h_nodeid),
4890			  le32_to_cpu(ms->m_lkid), saved_seq);
4891	} else if (error == -ENOENT) {
4892		log_error(ls, "receive %d no %x remote %d %x saved_seq %u",
4893			  le32_to_cpu(ms->m_type), le32_to_cpu(ms->m_remid),
4894			  le32_to_cpu(ms->m_header.h_nodeid),
4895			  le32_to_cpu(ms->m_lkid), saved_seq);
4896
4897		if (ms->m_type == cpu_to_le32(DLM_MSG_CONVERT))
4898			dlm_dump_rsb_hash(ls, le32_to_cpu(ms->m_hash));
4899	}
4900
4901	if (error == -EINVAL) {
4902		log_error(ls, "receive %d inval from %d lkid %x remid %x "
4903			  "saved_seq %u",
4904			  le32_to_cpu(ms->m_type),
4905			  le32_to_cpu(ms->m_header.h_nodeid),
4906			  le32_to_cpu(ms->m_lkid), le32_to_cpu(ms->m_remid),
4907			  saved_seq);
4908	}
4909}
4910
4911/* If the lockspace is in recovery mode (locking stopped), then normal
4912   messages are saved on the requestqueue for processing after recovery is
4913   done.  When not in recovery mode, we wait for dlm_recoverd to drain saved
4914   messages off the requestqueue before we process new ones. This occurs right
4915   after recovery completes when we transition from saving all messages on
4916   requestqueue, to processing all the saved messages, to processing new
4917   messages as they arrive. */
4918
4919static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms,
4920				int nodeid)
4921{
4922	if (dlm_locking_stopped(ls)) {
4923		/* If we were a member of this lockspace, left, and rejoined,
4924		   other nodes may still be sending us messages from the
4925		   lockspace generation before we left. */
4926		if (!ls->ls_generation) {
4927			log_limit(ls, "receive %d from %d ignore old gen",
4928				  le32_to_cpu(ms->m_type), nodeid);
4929			return;
4930		}
4931
4932		dlm_add_requestqueue(ls, nodeid, ms);
4933	} else {
4934		dlm_wait_requestqueue(ls);
4935		_receive_message(ls, ms, 0);
4936	}
4937}
4938
4939/* This is called by dlm_recoverd to process messages that were saved on
4940   the requestqueue. */
4941
4942void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms,
4943			       uint32_t saved_seq)
4944{
4945	_receive_message(ls, ms, saved_seq);
4946}
4947
4948/* This is called by the midcomms layer when something is received for
4949   the lockspace.  It could be either a MSG (normal message sent as part of
4950   standard locking activity) or an RCOM (recovery message sent as part of
4951   lockspace recovery). */
4952
4953void dlm_receive_buffer(union dlm_packet *p, int nodeid)
4954{
4955	struct dlm_header *hd = &p->header;
4956	struct dlm_ls *ls;
4957	int type = 0;
4958
4959	switch (hd->h_cmd) {
4960	case DLM_MSG:
4961		type = le32_to_cpu(p->message.m_type);
 
4962		break;
4963	case DLM_RCOM:
4964		type = le32_to_cpu(p->rcom.rc_type);
 
4965		break;
4966	default:
4967		log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid);
4968		return;
4969	}
4970
4971	if (le32_to_cpu(hd->h_nodeid) != nodeid) {
4972		log_print("invalid h_nodeid %d from %d lockspace %x",
4973			  le32_to_cpu(hd->h_nodeid), nodeid,
4974			  le32_to_cpu(hd->u.h_lockspace));
4975		return;
4976	}
4977
4978	ls = dlm_find_lockspace_global(le32_to_cpu(hd->u.h_lockspace));
4979	if (!ls) {
4980		if (dlm_config.ci_log_debug) {
4981			printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace "
4982				"%u from %d cmd %d type %d\n",
4983				le32_to_cpu(hd->u.h_lockspace), nodeid,
4984				hd->h_cmd, type);
4985		}
4986
4987		if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS)
4988			dlm_send_ls_not_ready(nodeid, &p->rcom);
4989		return;
4990	}
4991
4992	/* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to
4993	   be inactive (in this ls) before transitioning to recovery mode */
4994
4995	down_read(&ls->ls_recv_active);
4996	if (hd->h_cmd == DLM_MSG)
4997		dlm_receive_message(ls, &p->message, nodeid);
4998	else if (hd->h_cmd == DLM_RCOM)
4999		dlm_receive_rcom(ls, &p->rcom, nodeid);
5000	else
5001		log_error(ls, "invalid h_cmd %d from %d lockspace %x",
5002			  hd->h_cmd, nodeid, le32_to_cpu(hd->u.h_lockspace));
5003	up_read(&ls->ls_recv_active);
5004
5005	dlm_put_lockspace(ls);
5006}
5007
5008static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb,
5009				   struct dlm_message *ms_stub)
5010{
5011	if (middle_conversion(lkb)) {
5012		hold_lkb(lkb);
5013		memset(ms_stub, 0, sizeof(struct dlm_message));
5014		ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5015		ms_stub->m_type = cpu_to_le32(DLM_MSG_CONVERT_REPLY);
5016		ms_stub->m_result = cpu_to_le32(to_dlm_errno(-EINPROGRESS));
5017		ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5018		_receive_convert_reply(lkb, ms_stub);
5019
5020		/* Same special case as in receive_rcom_lock_args() */
5021		lkb->lkb_grmode = DLM_LOCK_IV;
5022		rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT);
5023		unhold_lkb(lkb);
5024
5025	} else if (lkb->lkb_rqmode >= lkb->lkb_grmode) {
5026		lkb->lkb_flags |= DLM_IFL_RESEND;
5027	}
5028
5029	/* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down
5030	   conversions are async; there's no reply from the remote master */
5031}
5032
5033/* A waiting lkb needs recovery if the master node has failed, or
5034   the master node is changing (only when no directory is used) */
5035
5036static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb,
5037				 int dir_nodeid)
5038{
5039	if (dlm_no_directory(ls))
5040		return 1;
5041
5042	if (dlm_is_removed(ls, lkb->lkb_wait_nodeid))
5043		return 1;
5044
5045	return 0;
5046}
5047
5048/* Recovery for locks that are waiting for replies from nodes that are now
5049   gone.  We can just complete unlocks and cancels by faking a reply from the
5050   dead node.  Requests and up-conversions we flag to be resent after
5051   recovery.  Down-conversions can just be completed with a fake reply like
5052   unlocks.  Conversions between PR and CW need special attention. */
5053
5054void dlm_recover_waiters_pre(struct dlm_ls *ls)
5055{
5056	struct dlm_lkb *lkb, *safe;
5057	struct dlm_message *ms_stub;
5058	int wait_type, stub_unlock_result, stub_cancel_result;
5059	int dir_nodeid;
5060
5061	ms_stub = kmalloc(sizeof(*ms_stub), GFP_KERNEL);
5062	if (!ms_stub)
 
5063		return;
 
5064
5065	mutex_lock(&ls->ls_waiters_mutex);
5066
5067	list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) {
5068
5069		dir_nodeid = dlm_dir_nodeid(lkb->lkb_resource);
5070
5071		/* exclude debug messages about unlocks because there can be so
5072		   many and they aren't very interesting */
5073
5074		if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) {
5075			log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5076				  "lkb_nodeid %d wait_nodeid %d dir_nodeid %d",
5077				  lkb->lkb_id,
5078				  lkb->lkb_remid,
5079				  lkb->lkb_wait_type,
5080				  lkb->lkb_resource->res_nodeid,
5081				  lkb->lkb_nodeid,
5082				  lkb->lkb_wait_nodeid,
5083				  dir_nodeid);
5084		}
5085
5086		/* all outstanding lookups, regardless of destination  will be
5087		   resent after recovery is done */
5088
5089		if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) {
5090			lkb->lkb_flags |= DLM_IFL_RESEND;
5091			continue;
5092		}
5093
5094		if (!waiter_needs_recovery(ls, lkb, dir_nodeid))
5095			continue;
5096
5097		wait_type = lkb->lkb_wait_type;
5098		stub_unlock_result = -DLM_EUNLOCK;
5099		stub_cancel_result = -DLM_ECANCEL;
5100
5101		/* Main reply may have been received leaving a zero wait_type,
5102		   but a reply for the overlapping op may not have been
5103		   received.  In that case we need to fake the appropriate
5104		   reply for the overlap op. */
5105
5106		if (!wait_type) {
5107			if (is_overlap_cancel(lkb)) {
5108				wait_type = DLM_MSG_CANCEL;
5109				if (lkb->lkb_grmode == DLM_LOCK_IV)
5110					stub_cancel_result = 0;
5111			}
5112			if (is_overlap_unlock(lkb)) {
5113				wait_type = DLM_MSG_UNLOCK;
5114				if (lkb->lkb_grmode == DLM_LOCK_IV)
5115					stub_unlock_result = -ENOENT;
5116			}
5117
5118			log_debug(ls, "rwpre overlap %x %x %d %d %d",
5119				  lkb->lkb_id, lkb->lkb_flags, wait_type,
5120				  stub_cancel_result, stub_unlock_result);
5121		}
5122
5123		switch (wait_type) {
5124
5125		case DLM_MSG_REQUEST:
5126			lkb->lkb_flags |= DLM_IFL_RESEND;
5127			break;
5128
5129		case DLM_MSG_CONVERT:
5130			recover_convert_waiter(ls, lkb, ms_stub);
5131			break;
5132
5133		case DLM_MSG_UNLOCK:
5134			hold_lkb(lkb);
5135			memset(ms_stub, 0, sizeof(struct dlm_message));
5136			ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5137			ms_stub->m_type = cpu_to_le32(DLM_MSG_UNLOCK_REPLY);
5138			ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_unlock_result));
5139			ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5140			_receive_unlock_reply(lkb, ms_stub);
5141			dlm_put_lkb(lkb);
5142			break;
5143
5144		case DLM_MSG_CANCEL:
5145			hold_lkb(lkb);
5146			memset(ms_stub, 0, sizeof(struct dlm_message));
5147			ms_stub->m_flags = cpu_to_le32(DLM_IFL_STUB_MS);
5148			ms_stub->m_type = cpu_to_le32(DLM_MSG_CANCEL_REPLY);
5149			ms_stub->m_result = cpu_to_le32(to_dlm_errno(stub_cancel_result));
5150			ms_stub->m_header.h_nodeid = cpu_to_le32(lkb->lkb_nodeid);
5151			_receive_cancel_reply(lkb, ms_stub);
5152			dlm_put_lkb(lkb);
5153			break;
5154
5155		default:
5156			log_error(ls, "invalid lkb wait_type %d %d",
5157				  lkb->lkb_wait_type, wait_type);
5158		}
5159		schedule();
5160	}
5161	mutex_unlock(&ls->ls_waiters_mutex);
5162	kfree(ms_stub);
5163}
5164
5165static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls)
5166{
5167	struct dlm_lkb *lkb = NULL, *iter;
 
5168
5169	mutex_lock(&ls->ls_waiters_mutex);
5170	list_for_each_entry(iter, &ls->ls_waiters, lkb_wait_reply) {
5171		if (iter->lkb_flags & DLM_IFL_RESEND) {
5172			hold_lkb(iter);
5173			lkb = iter;
5174			break;
5175		}
5176	}
5177	mutex_unlock(&ls->ls_waiters_mutex);
5178
 
 
5179	return lkb;
5180}
5181
5182/* Deal with lookups and lkb's marked RESEND from _pre.  We may now be the
5183   master or dir-node for r.  Processing the lkb may result in it being placed
5184   back on waiters. */
5185
5186/* We do this after normal locking has been enabled and any saved messages
5187   (in requestqueue) have been processed.  We should be confident that at
5188   this point we won't get or process a reply to any of these waiting
5189   operations.  But, new ops may be coming in on the rsbs/locks here from
5190   userspace or remotely. */
5191
5192/* there may have been an overlap unlock/cancel prior to recovery or after
5193   recovery.  if before, the lkb may still have a pos wait_count; if after, the
5194   overlap flag would just have been set and nothing new sent.  we can be
5195   confident here than any replies to either the initial op or overlap ops
5196   prior to recovery have been received. */
5197
5198int dlm_recover_waiters_post(struct dlm_ls *ls)
5199{
5200	struct dlm_lkb *lkb;
5201	struct dlm_rsb *r;
5202	int error = 0, mstype, err, oc, ou;
5203
5204	while (1) {
5205		if (dlm_locking_stopped(ls)) {
5206			log_debug(ls, "recover_waiters_post aborted");
5207			error = -EINTR;
5208			break;
5209		}
5210
5211		lkb = find_resend_waiter(ls);
5212		if (!lkb)
5213			break;
5214
5215		r = lkb->lkb_resource;
5216		hold_rsb(r);
5217		lock_rsb(r);
5218
5219		mstype = lkb->lkb_wait_type;
5220		oc = is_overlap_cancel(lkb);
5221		ou = is_overlap_unlock(lkb);
5222		err = 0;
5223
5224		log_debug(ls, "waiter %x remote %x msg %d r_nodeid %d "
5225			  "lkb_nodeid %d wait_nodeid %d dir_nodeid %d "
5226			  "overlap %d %d", lkb->lkb_id, lkb->lkb_remid, mstype,
5227			  r->res_nodeid, lkb->lkb_nodeid, lkb->lkb_wait_nodeid,
5228			  dlm_dir_nodeid(r), oc, ou);
5229
5230		/* At this point we assume that we won't get a reply to any
5231		   previous op or overlap op on this lock.  First, do a big
5232		   remove_from_waiters() for all previous ops. */
5233
5234		lkb->lkb_flags &= ~DLM_IFL_RESEND;
5235		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK;
5236		lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL;
5237		lkb->lkb_wait_type = 0;
5238		/* drop all wait_count references we still
5239		 * hold a reference for this iteration.
5240		 */
5241		while (lkb->lkb_wait_count) {
5242			lkb->lkb_wait_count--;
5243			unhold_lkb(lkb);
5244		}
5245		mutex_lock(&ls->ls_waiters_mutex);
5246		list_del_init(&lkb->lkb_wait_reply);
5247		mutex_unlock(&ls->ls_waiters_mutex);
 
5248
5249		if (oc || ou) {
5250			/* do an unlock or cancel instead of resending */
5251			switch (mstype) {
5252			case DLM_MSG_LOOKUP:
5253			case DLM_MSG_REQUEST:
5254				queue_cast(r, lkb, ou ? -DLM_EUNLOCK :
5255							-DLM_ECANCEL);
5256				unhold_lkb(lkb); /* undoes create_lkb() */
5257				break;
5258			case DLM_MSG_CONVERT:
5259				if (oc) {
5260					queue_cast(r, lkb, -DLM_ECANCEL);
5261				} else {
5262					lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK;
5263					_unlock_lock(r, lkb);
5264				}
5265				break;
5266			default:
5267				err = 1;
5268			}
5269		} else {
5270			switch (mstype) {
5271			case DLM_MSG_LOOKUP:
5272			case DLM_MSG_REQUEST:
5273				_request_lock(r, lkb);
5274				if (is_master(r))
5275					confirm_master(r, 0);
5276				break;
5277			case DLM_MSG_CONVERT:
5278				_convert_lock(r, lkb);
5279				break;
5280			default:
5281				err = 1;
5282			}
5283		}
5284
5285		if (err) {
5286			log_error(ls, "waiter %x msg %d r_nodeid %d "
5287				  "dir_nodeid %d overlap %d %d",
5288				  lkb->lkb_id, mstype, r->res_nodeid,
5289				  dlm_dir_nodeid(r), oc, ou);
5290		}
5291		unlock_rsb(r);
5292		put_rsb(r);
5293		dlm_put_lkb(lkb);
5294	}
5295
5296	return error;
5297}
5298
5299static void purge_mstcpy_list(struct dlm_ls *ls, struct dlm_rsb *r,
5300			      struct list_head *list)
5301{
5302	struct dlm_lkb *lkb, *safe;
5303
5304	list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5305		if (!is_master_copy(lkb))
5306			continue;
5307
5308		/* don't purge lkbs we've added in recover_master_copy for
5309		   the current recovery seq */
5310
5311		if (lkb->lkb_recover_seq == ls->ls_recover_seq)
5312			continue;
5313
5314		del_lkb(r, lkb);
5315
5316		/* this put should free the lkb */
5317		if (!dlm_put_lkb(lkb))
5318			log_error(ls, "purged mstcpy lkb not released");
5319	}
5320}
5321
5322void dlm_purge_mstcpy_locks(struct dlm_rsb *r)
5323{
5324	struct dlm_ls *ls = r->res_ls;
5325
5326	purge_mstcpy_list(ls, r, &r->res_grantqueue);
5327	purge_mstcpy_list(ls, r, &r->res_convertqueue);
5328	purge_mstcpy_list(ls, r, &r->res_waitqueue);
5329}
5330
5331static void purge_dead_list(struct dlm_ls *ls, struct dlm_rsb *r,
5332			    struct list_head *list,
5333			    int nodeid_gone, unsigned int *count)
5334{
5335	struct dlm_lkb *lkb, *safe;
5336
5337	list_for_each_entry_safe(lkb, safe, list, lkb_statequeue) {
5338		if (!is_master_copy(lkb))
5339			continue;
5340
5341		if ((lkb->lkb_nodeid == nodeid_gone) ||
5342		    dlm_is_removed(ls, lkb->lkb_nodeid)) {
5343
5344			/* tell recover_lvb to invalidate the lvb
5345			   because a node holding EX/PW failed */
5346			if ((lkb->lkb_exflags & DLM_LKF_VALBLK) &&
5347			    (lkb->lkb_grmode >= DLM_LOCK_PW)) {
5348				rsb_set_flag(r, RSB_RECOVER_LVB_INVAL);
5349			}
5350
5351			del_lkb(r, lkb);
5352
5353			/* this put should free the lkb */
5354			if (!dlm_put_lkb(lkb))
5355				log_error(ls, "purged dead lkb not released");
5356
5357			rsb_set_flag(r, RSB_RECOVER_GRANT);
5358
5359			(*count)++;
5360		}
5361	}
5362}
5363
5364/* Get rid of locks held by nodes that are gone. */
5365
5366void dlm_recover_purge(struct dlm_ls *ls)
5367{
5368	struct dlm_rsb *r;
5369	struct dlm_member *memb;
5370	int nodes_count = 0;
5371	int nodeid_gone = 0;
5372	unsigned int lkb_count = 0;
5373
5374	/* cache one removed nodeid to optimize the common
5375	   case of a single node removed */
5376
5377	list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
5378		nodes_count++;
5379		nodeid_gone = memb->nodeid;
5380	}
5381
5382	if (!nodes_count)
5383		return;
5384
5385	down_write(&ls->ls_root_sem);
5386	list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
5387		hold_rsb(r);
5388		lock_rsb(r);
5389		if (is_master(r)) {
5390			purge_dead_list(ls, r, &r->res_grantqueue,
5391					nodeid_gone, &lkb_count);
5392			purge_dead_list(ls, r, &r->res_convertqueue,
5393					nodeid_gone, &lkb_count);
5394			purge_dead_list(ls, r, &r->res_waitqueue,
5395					nodeid_gone, &lkb_count);
5396		}
5397		unlock_rsb(r);
5398		unhold_rsb(r);
5399		cond_resched();
5400	}
5401	up_write(&ls->ls_root_sem);
5402
5403	if (lkb_count)
5404		log_rinfo(ls, "dlm_recover_purge %u locks for %u nodes",
5405			  lkb_count, nodes_count);
5406}
5407
5408static struct dlm_rsb *find_grant_rsb(struct dlm_ls *ls, int bucket)
5409{
5410	struct rb_node *n;
5411	struct dlm_rsb *r;
5412
5413	spin_lock(&ls->ls_rsbtbl[bucket].lock);
5414	for (n = rb_first(&ls->ls_rsbtbl[bucket].keep); n; n = rb_next(n)) {
5415		r = rb_entry(n, struct dlm_rsb, res_hashnode);
5416
5417		if (!rsb_flag(r, RSB_RECOVER_GRANT))
5418			continue;
5419		if (!is_master(r)) {
5420			rsb_clear_flag(r, RSB_RECOVER_GRANT);
5421			continue;
5422		}
5423		hold_rsb(r);
5424		spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5425		return r;
5426	}
5427	spin_unlock(&ls->ls_rsbtbl[bucket].lock);
5428	return NULL;
5429}
5430
5431/*
5432 * Attempt to grant locks on resources that we are the master of.
5433 * Locks may have become grantable during recovery because locks
5434 * from departed nodes have been purged (or not rebuilt), allowing
5435 * previously blocked locks to now be granted.  The subset of rsb's
5436 * we are interested in are those with lkb's on either the convert or
5437 * waiting queues.
5438 *
5439 * Simplest would be to go through each master rsb and check for non-empty
5440 * convert or waiting queues, and attempt to grant on those rsbs.
5441 * Checking the queues requires lock_rsb, though, for which we'd need
5442 * to release the rsbtbl lock.  This would make iterating through all
5443 * rsb's very inefficient.  So, we rely on earlier recovery routines
5444 * to set RECOVER_GRANT on any rsb's that we should attempt to grant
5445 * locks for.
5446 */
5447
5448void dlm_recover_grant(struct dlm_ls *ls)
5449{
5450	struct dlm_rsb *r;
5451	int bucket = 0;
5452	unsigned int count = 0;
5453	unsigned int rsb_count = 0;
5454	unsigned int lkb_count = 0;
5455
5456	while (1) {
5457		r = find_grant_rsb(ls, bucket);
5458		if (!r) {
5459			if (bucket == ls->ls_rsbtbl_size - 1)
5460				break;
5461			bucket++;
5462			continue;
5463		}
5464		rsb_count++;
5465		count = 0;
5466		lock_rsb(r);
5467		/* the RECOVER_GRANT flag is checked in the grant path */
5468		grant_pending_locks(r, &count);
5469		rsb_clear_flag(r, RSB_RECOVER_GRANT);
5470		lkb_count += count;
5471		confirm_master(r, 0);
5472		unlock_rsb(r);
5473		put_rsb(r);
5474		cond_resched();
5475	}
5476
5477	if (lkb_count)
5478		log_rinfo(ls, "dlm_recover_grant %u locks on %u resources",
5479			  lkb_count, rsb_count);
5480}
5481
5482static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid,
5483					 uint32_t remid)
5484{
5485	struct dlm_lkb *lkb;
5486
5487	list_for_each_entry(lkb, head, lkb_statequeue) {
5488		if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid)
5489			return lkb;
5490	}
5491	return NULL;
5492}
5493
5494static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid,
5495				    uint32_t remid)
5496{
5497	struct dlm_lkb *lkb;
5498
5499	lkb = search_remid_list(&r->res_grantqueue, nodeid, remid);
5500	if (lkb)
5501		return lkb;
5502	lkb = search_remid_list(&r->res_convertqueue, nodeid, remid);
5503	if (lkb)
5504		return lkb;
5505	lkb = search_remid_list(&r->res_waitqueue, nodeid, remid);
5506	if (lkb)
5507		return lkb;
5508	return NULL;
5509}
5510
5511/* needs at least dlm_rcom + rcom_lock */
5512static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb,
5513				  struct dlm_rsb *r, struct dlm_rcom *rc)
5514{
5515	struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5516
5517	lkb->lkb_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
5518	lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid);
5519	lkb->lkb_remid = le32_to_cpu(rl->rl_lkid);
5520	lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags);
5521	lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF;
5522	lkb->lkb_flags |= DLM_IFL_MSTCPY;
5523	lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq);
5524	lkb->lkb_rqmode = rl->rl_rqmode;
5525	lkb->lkb_grmode = rl->rl_grmode;
5526	/* don't set lkb_status because add_lkb wants to itself */
5527
5528	lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL;
5529	lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL;
5530
5531	if (lkb->lkb_exflags & DLM_LKF_VALBLK) {
5532		int lvblen = le16_to_cpu(rc->rc_header.h_length) -
5533			sizeof(struct dlm_rcom) - sizeof(struct rcom_lock);
5534		if (lvblen > ls->ls_lvblen)
5535			return -EINVAL;
5536		lkb->lkb_lvbptr = dlm_allocate_lvb(ls);
5537		if (!lkb->lkb_lvbptr)
5538			return -ENOMEM;
5539		memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen);
5540	}
5541
5542	/* Conversions between PR and CW (middle modes) need special handling.
5543	   The real granted mode of these converting locks cannot be determined
5544	   until all locks have been rebuilt on the rsb (recover_conversion) */
5545
5546	if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) &&
5547	    middle_conversion(lkb)) {
5548		rl->rl_status = DLM_LKSTS_CONVERT;
5549		lkb->lkb_grmode = DLM_LOCK_IV;
5550		rsb_set_flag(r, RSB_RECOVER_CONVERT);
5551	}
5552
5553	return 0;
5554}
5555
5556/* This lkb may have been recovered in a previous aborted recovery so we need
5557   to check if the rsb already has an lkb with the given remote nodeid/lkid.
5558   If so we just send back a standard reply.  If not, we create a new lkb with
5559   the given values and send back our lkid.  We send back our lkid by sending
5560   back the rcom_lock struct we got but with the remid field filled in. */
5561
5562/* needs at least dlm_rcom + rcom_lock */
5563int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5564{
5565	struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5566	struct dlm_rsb *r;
5567	struct dlm_lkb *lkb;
5568	uint32_t remid = 0;
5569	int from_nodeid = le32_to_cpu(rc->rc_header.h_nodeid);
5570	int error;
5571
5572	if (rl->rl_parent_lkid) {
5573		error = -EOPNOTSUPP;
5574		goto out;
5575	}
5576
5577	remid = le32_to_cpu(rl->rl_lkid);
5578
5579	/* In general we expect the rsb returned to be R_MASTER, but we don't
5580	   have to require it.  Recovery of masters on one node can overlap
5581	   recovery of locks on another node, so one node can send us MSTCPY
5582	   locks before we've made ourselves master of this rsb.  We can still
5583	   add new MSTCPY locks that we receive here without any harm; when
5584	   we make ourselves master, dlm_recover_masters() won't touch the
5585	   MSTCPY locks we've received early. */
5586
5587	error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen),
5588			 from_nodeid, R_RECEIVE_RECOVER, &r);
5589	if (error)
5590		goto out;
5591
5592	lock_rsb(r);
5593
5594	if (dlm_no_directory(ls) && (dlm_dir_nodeid(r) != dlm_our_nodeid())) {
5595		log_error(ls, "dlm_recover_master_copy remote %d %x not dir",
5596			  from_nodeid, remid);
5597		error = -EBADR;
5598		goto out_unlock;
 
5599	}
5600
5601	lkb = search_remid(r, from_nodeid, remid);
 
 
5602	if (lkb) {
5603		error = -EEXIST;
5604		goto out_remid;
5605	}
5606
5607	error = create_lkb(ls, &lkb);
5608	if (error)
5609		goto out_unlock;
5610
5611	error = receive_rcom_lock_args(ls, lkb, r, rc);
5612	if (error) {
5613		__put_lkb(ls, lkb);
5614		goto out_unlock;
5615	}
5616
5617	attach_lkb(r, lkb);
5618	add_lkb(r, lkb, rl->rl_status);
 
5619	ls->ls_recover_locks_in++;
5620
5621	if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
5622		rsb_set_flag(r, RSB_RECOVER_GRANT);
5623
5624 out_remid:
5625	/* this is the new value returned to the lock holder for
5626	   saving in its process-copy lkb */
5627	rl->rl_remid = cpu_to_le32(lkb->lkb_id);
5628
5629	lkb->lkb_recover_seq = ls->ls_recover_seq;
5630
5631 out_unlock:
5632	unlock_rsb(r);
5633	put_rsb(r);
5634 out:
5635	if (error && error != -EEXIST)
5636		log_rinfo(ls, "dlm_recover_master_copy remote %d %x error %d",
5637			  from_nodeid, remid, error);
5638	rl->rl_result = cpu_to_le32(error);
5639	return error;
5640}
5641
5642/* needs at least dlm_rcom + rcom_lock */
5643int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc)
5644{
5645	struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf;
5646	struct dlm_rsb *r;
5647	struct dlm_lkb *lkb;
5648	uint32_t lkid, remid;
5649	int error, result;
5650
5651	lkid = le32_to_cpu(rl->rl_lkid);
5652	remid = le32_to_cpu(rl->rl_remid);
5653	result = le32_to_cpu(rl->rl_result);
5654
5655	error = find_lkb(ls, lkid, &lkb);
5656	if (error) {
5657		log_error(ls, "dlm_recover_process_copy no %x remote %d %x %d",
5658			  lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5659			  result);
5660		return error;
5661	}
5662
5663	r = lkb->lkb_resource;
5664	hold_rsb(r);
5665	lock_rsb(r);
5666
5667	if (!is_process_copy(lkb)) {
5668		log_error(ls, "dlm_recover_process_copy bad %x remote %d %x %d",
5669			  lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5670			  result);
5671		dlm_dump_rsb(r);
5672		unlock_rsb(r);
5673		put_rsb(r);
5674		dlm_put_lkb(lkb);
5675		return -EINVAL;
5676	}
5677
5678	switch (result) {
5679	case -EBADR:
5680		/* There's a chance the new master received our lock before
5681		   dlm_recover_master_reply(), this wouldn't happen if we did
5682		   a barrier between recover_masters and recover_locks. */
5683
5684		log_debug(ls, "dlm_recover_process_copy %x remote %d %x %d",
5685			  lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5686			  result);
5687	
5688		dlm_send_rcom_lock(r, lkb);
5689		goto out;
5690	case -EEXIST:
5691	case 0:
5692		lkb->lkb_remid = remid;
5693		break;
5694	default:
5695		log_error(ls, "dlm_recover_process_copy %x remote %d %x %d unk",
5696			  lkid, le32_to_cpu(rc->rc_header.h_nodeid), remid,
5697			  result);
5698	}
5699
5700	/* an ack for dlm_recover_locks() which waits for replies from
5701	   all the locks it sends to new masters */
5702	dlm_recovered_lock(r);
5703 out:
5704	unlock_rsb(r);
5705	put_rsb(r);
5706	dlm_put_lkb(lkb);
5707
5708	return 0;
5709}
5710
5711#ifdef CONFIG_DLM_DEPRECATED_API
5712int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5713		     int mode, uint32_t flags, void *name, unsigned int namelen,
5714		     unsigned long timeout_cs)
5715#else
5716int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua,
5717		     int mode, uint32_t flags, void *name, unsigned int namelen)
5718#endif
5719{
5720	struct dlm_lkb *lkb;
5721	struct dlm_args args;
5722	bool do_put = true;
5723	int error;
5724
5725	dlm_lock_recovery(ls);
5726
5727	error = create_lkb(ls, &lkb);
5728	if (error) {
5729		kfree(ua);
5730		goto out;
5731	}
5732
5733	trace_dlm_lock_start(ls, lkb, name, namelen, mode, flags);
5734
5735	if (flags & DLM_LKF_VALBLK) {
5736		ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5737		if (!ua->lksb.sb_lvbptr) {
5738			kfree(ua);
 
5739			error = -ENOMEM;
5740			goto out_put;
5741		}
5742	}
5743#ifdef CONFIG_DLM_DEPRECATED_API
5744	error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs,
5745			      fake_astfn, ua, fake_bastfn, &args);
5746#else
5747	error = set_lock_args(mode, &ua->lksb, flags, namelen, fake_astfn, ua,
5748			      fake_bastfn, &args);
5749#endif
5750	if (error) {
5751		kfree(ua->lksb.sb_lvbptr);
5752		ua->lksb.sb_lvbptr = NULL;
5753		kfree(ua);
5754		goto out_put;
5755	}
5756
5757	/* After ua is attached to lkb it will be freed by dlm_free_lkb().
5758	   When DLM_IFL_USER is set, the dlm knows that this is a userspace
5759	   lock and that lkb_astparam is the dlm_user_args structure. */
 
 
 
5760	lkb->lkb_flags |= DLM_IFL_USER;
 
 
 
 
 
 
5761	error = request_lock(ls, lkb, name, namelen, &args);
5762
5763	switch (error) {
5764	case 0:
5765		break;
5766	case -EINPROGRESS:
5767		error = 0;
5768		break;
5769	case -EAGAIN:
5770		error = 0;
5771		fallthrough;
5772	default:
5773		goto out_put;
 
5774	}
5775
5776	/* add this new lkb to the per-process list of locks */
5777	spin_lock(&ua->proc->locks_spin);
5778	hold_lkb(lkb);
5779	list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5780	spin_unlock(&ua->proc->locks_spin);
5781	do_put = false;
5782 out_put:
5783	trace_dlm_lock_end(ls, lkb, name, namelen, mode, flags, error, false);
5784	if (do_put)
5785		__put_lkb(ls, lkb);
5786 out:
5787	dlm_unlock_recovery(ls);
5788	return error;
5789}
5790
5791#ifdef CONFIG_DLM_DEPRECATED_API
5792int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5793		     int mode, uint32_t flags, uint32_t lkid, char *lvb_in,
5794		     unsigned long timeout_cs)
5795#else
5796int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5797		     int mode, uint32_t flags, uint32_t lkid, char *lvb_in)
5798#endif
5799{
5800	struct dlm_lkb *lkb;
5801	struct dlm_args args;
5802	struct dlm_user_args *ua;
5803	int error;
5804
5805	dlm_lock_recovery(ls);
5806
5807	error = find_lkb(ls, lkid, &lkb);
5808	if (error)
5809		goto out;
5810
5811	trace_dlm_lock_start(ls, lkb, NULL, 0, mode, flags);
5812
5813	/* user can change the params on its lock when it converts it, or
5814	   add an lvb that didn't exist before */
5815
5816	ua = lkb->lkb_ua;
5817
5818	if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) {
5819		ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS);
5820		if (!ua->lksb.sb_lvbptr) {
5821			error = -ENOMEM;
5822			goto out_put;
5823		}
5824	}
5825	if (lvb_in && ua->lksb.sb_lvbptr)
5826		memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5827
5828	ua->xid = ua_tmp->xid;
5829	ua->castparam = ua_tmp->castparam;
5830	ua->castaddr = ua_tmp->castaddr;
5831	ua->bastparam = ua_tmp->bastparam;
5832	ua->bastaddr = ua_tmp->bastaddr;
5833	ua->user_lksb = ua_tmp->user_lksb;
5834
5835#ifdef CONFIG_DLM_DEPRECATED_API
5836	error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs,
5837			      fake_astfn, ua, fake_bastfn, &args);
5838#else
5839	error = set_lock_args(mode, &ua->lksb, flags, 0, fake_astfn, ua,
5840			      fake_bastfn, &args);
5841#endif
5842	if (error)
5843		goto out_put;
5844
5845	error = convert_lock(ls, lkb, &args);
5846
5847	if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK)
5848		error = 0;
5849 out_put:
5850	trace_dlm_lock_end(ls, lkb, NULL, 0, mode, flags, error, false);
5851	dlm_put_lkb(lkb);
5852 out:
5853	dlm_unlock_recovery(ls);
5854	kfree(ua_tmp);
5855	return error;
5856}
5857
5858/*
5859 * The caller asks for an orphan lock on a given resource with a given mode.
5860 * If a matching lock exists, it's moved to the owner's list of locks and
5861 * the lkid is returned.
5862 */
5863
5864int dlm_user_adopt_orphan(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5865		     int mode, uint32_t flags, void *name, unsigned int namelen,
5866		     uint32_t *lkid)
5867{
5868	struct dlm_lkb *lkb = NULL, *iter;
5869	struct dlm_user_args *ua;
5870	int found_other_mode = 0;
5871	int rv = 0;
5872
5873	mutex_lock(&ls->ls_orphans_mutex);
5874	list_for_each_entry(iter, &ls->ls_orphans, lkb_ownqueue) {
5875		if (iter->lkb_resource->res_length != namelen)
5876			continue;
5877		if (memcmp(iter->lkb_resource->res_name, name, namelen))
5878			continue;
5879		if (iter->lkb_grmode != mode) {
5880			found_other_mode = 1;
5881			continue;
5882		}
5883
5884		lkb = iter;
5885		list_del_init(&iter->lkb_ownqueue);
5886		iter->lkb_flags &= ~DLM_IFL_ORPHAN;
5887		*lkid = iter->lkb_id;
5888		break;
5889	}
5890	mutex_unlock(&ls->ls_orphans_mutex);
5891
5892	if (!lkb && found_other_mode) {
5893		rv = -EAGAIN;
5894		goto out;
5895	}
5896
5897	if (!lkb) {
5898		rv = -ENOENT;
5899		goto out;
5900	}
5901
5902	lkb->lkb_exflags = flags;
5903	lkb->lkb_ownpid = (int) current->pid;
5904
5905	ua = lkb->lkb_ua;
5906
5907	ua->proc = ua_tmp->proc;
5908	ua->xid = ua_tmp->xid;
5909	ua->castparam = ua_tmp->castparam;
5910	ua->castaddr = ua_tmp->castaddr;
5911	ua->bastparam = ua_tmp->bastparam;
5912	ua->bastaddr = ua_tmp->bastaddr;
5913	ua->user_lksb = ua_tmp->user_lksb;
5914
5915	/*
5916	 * The lkb reference from the ls_orphans list was not
5917	 * removed above, and is now considered the reference
5918	 * for the proc locks list.
5919	 */
5920
5921	spin_lock(&ua->proc->locks_spin);
5922	list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks);
5923	spin_unlock(&ua->proc->locks_spin);
5924 out:
5925	kfree(ua_tmp);
5926	return rv;
5927}
5928
5929int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5930		    uint32_t flags, uint32_t lkid, char *lvb_in)
5931{
5932	struct dlm_lkb *lkb;
5933	struct dlm_args args;
5934	struct dlm_user_args *ua;
5935	int error;
5936
5937	dlm_lock_recovery(ls);
5938
5939	error = find_lkb(ls, lkid, &lkb);
5940	if (error)
5941		goto out;
5942
5943	trace_dlm_unlock_start(ls, lkb, flags);
5944
5945	ua = lkb->lkb_ua;
5946
5947	if (lvb_in && ua->lksb.sb_lvbptr)
5948		memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN);
5949	if (ua_tmp->castparam)
5950		ua->castparam = ua_tmp->castparam;
5951	ua->user_lksb = ua_tmp->user_lksb;
5952
5953	error = set_unlock_args(flags, ua, &args);
5954	if (error)
5955		goto out_put;
5956
5957	error = unlock_lock(ls, lkb, &args);
5958
5959	if (error == -DLM_EUNLOCK)
5960		error = 0;
5961	/* from validate_unlock_args() */
5962	if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK))
5963		error = 0;
5964	if (error)
5965		goto out_put;
5966
5967	spin_lock(&ua->proc->locks_spin);
5968	/* dlm_user_add_cb() may have already taken lkb off the proc list */
5969	if (!list_empty(&lkb->lkb_ownqueue))
5970		list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking);
5971	spin_unlock(&ua->proc->locks_spin);
5972 out_put:
5973	trace_dlm_unlock_end(ls, lkb, flags, error);
5974	dlm_put_lkb(lkb);
5975 out:
5976	dlm_unlock_recovery(ls);
5977	kfree(ua_tmp);
5978	return error;
5979}
5980
5981int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp,
5982		    uint32_t flags, uint32_t lkid)
5983{
5984	struct dlm_lkb *lkb;
5985	struct dlm_args args;
5986	struct dlm_user_args *ua;
5987	int error;
5988
5989	dlm_lock_recovery(ls);
5990
5991	error = find_lkb(ls, lkid, &lkb);
5992	if (error)
5993		goto out;
5994
5995	trace_dlm_unlock_start(ls, lkb, flags);
5996
5997	ua = lkb->lkb_ua;
5998	if (ua_tmp->castparam)
5999		ua->castparam = ua_tmp->castparam;
6000	ua->user_lksb = ua_tmp->user_lksb;
6001
6002	error = set_unlock_args(flags, ua, &args);
6003	if (error)
6004		goto out_put;
6005
6006	error = cancel_lock(ls, lkb, &args);
6007
6008	if (error == -DLM_ECANCEL)
6009		error = 0;
6010	/* from validate_unlock_args() */
6011	if (error == -EBUSY)
6012		error = 0;
6013 out_put:
6014	trace_dlm_unlock_end(ls, lkb, flags, error);
6015	dlm_put_lkb(lkb);
6016 out:
6017	dlm_unlock_recovery(ls);
6018	kfree(ua_tmp);
6019	return error;
6020}
6021
6022int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid)
6023{
6024	struct dlm_lkb *lkb;
6025	struct dlm_args args;
6026	struct dlm_user_args *ua;
6027	struct dlm_rsb *r;
6028	int error;
6029
6030	dlm_lock_recovery(ls);
6031
6032	error = find_lkb(ls, lkid, &lkb);
6033	if (error)
6034		goto out;
6035
6036	trace_dlm_unlock_start(ls, lkb, flags);
6037
6038	ua = lkb->lkb_ua;
6039
6040	error = set_unlock_args(flags, ua, &args);
6041	if (error)
6042		goto out_put;
6043
6044	/* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */
6045
6046	r = lkb->lkb_resource;
6047	hold_rsb(r);
6048	lock_rsb(r);
6049
6050	error = validate_unlock_args(lkb, &args);
6051	if (error)
6052		goto out_r;
6053	lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL;
6054
6055	error = _cancel_lock(r, lkb);
6056 out_r:
6057	unlock_rsb(r);
6058	put_rsb(r);
6059
6060	if (error == -DLM_ECANCEL)
6061		error = 0;
6062	/* from validate_unlock_args() */
6063	if (error == -EBUSY)
6064		error = 0;
6065 out_put:
6066	trace_dlm_unlock_end(ls, lkb, flags, error);
6067	dlm_put_lkb(lkb);
6068 out:
6069	dlm_unlock_recovery(ls);
6070	return error;
6071}
6072
6073/* lkb's that are removed from the waiters list by revert are just left on the
6074   orphans list with the granted orphan locks, to be freed by purge */
6075
6076static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6077{
6078	struct dlm_args args;
6079	int error;
6080
6081	hold_lkb(lkb); /* reference for the ls_orphans list */
6082	mutex_lock(&ls->ls_orphans_mutex);
6083	list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans);
6084	mutex_unlock(&ls->ls_orphans_mutex);
6085
6086	set_unlock_args(0, lkb->lkb_ua, &args);
6087
6088	error = cancel_lock(ls, lkb, &args);
6089	if (error == -DLM_ECANCEL)
6090		error = 0;
6091	return error;
6092}
6093
6094/* The FORCEUNLOCK flag allows the unlock to go ahead even if the lkb isn't
6095   granted.  Regardless of what rsb queue the lock is on, it's removed and
6096   freed.  The IVVALBLK flag causes the lvb on the resource to be invalidated
6097   if our lock is PW/EX (it's ignored if our granted mode is smaller.) */
6098
6099static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb)
6100{
6101	struct dlm_args args;
6102	int error;
6103
6104	set_unlock_args(DLM_LKF_FORCEUNLOCK | DLM_LKF_IVVALBLK,
6105			lkb->lkb_ua, &args);
6106
6107	error = unlock_lock(ls, lkb, &args);
6108	if (error == -DLM_EUNLOCK)
6109		error = 0;
6110	return error;
6111}
6112
6113/* We have to release clear_proc_locks mutex before calling unlock_proc_lock()
6114   (which does lock_rsb) due to deadlock with receiving a message that does
6115   lock_rsb followed by dlm_user_add_cb() */
6116
6117static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls,
6118				     struct dlm_user_proc *proc)
6119{
6120	struct dlm_lkb *lkb = NULL;
6121
6122	spin_lock(&ls->ls_clear_proc_locks);
6123	if (list_empty(&proc->locks))
6124		goto out;
6125
6126	lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue);
6127	list_del_init(&lkb->lkb_ownqueue);
6128
6129	if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6130		lkb->lkb_flags |= DLM_IFL_ORPHAN;
6131	else
6132		lkb->lkb_flags |= DLM_IFL_DEAD;
6133 out:
6134	spin_unlock(&ls->ls_clear_proc_locks);
6135	return lkb;
6136}
6137
6138/* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which
6139   1) references lkb->ua which we free here and 2) adds lkbs to proc->asts,
6140   which we clear here. */
6141
6142/* proc CLOSING flag is set so no more device_reads should look at proc->asts
6143   list, and no more device_writes should add lkb's to proc->locks list; so we
6144   shouldn't need to take asts_spin or locks_spin here.  this assumes that
6145   device reads/writes/closes are serialized -- FIXME: we may need to serialize
6146   them ourself. */
6147
6148void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6149{
6150	struct dlm_lkb *lkb, *safe;
6151
6152	dlm_lock_recovery(ls);
6153
6154	while (1) {
6155		lkb = del_proc_lock(ls, proc);
6156		if (!lkb)
6157			break;
6158		del_timeout(lkb);
6159		if (lkb->lkb_exflags & DLM_LKF_PERSISTENT)
6160			orphan_proc_lock(ls, lkb);
6161		else
6162			unlock_proc_lock(ls, lkb);
6163
6164		/* this removes the reference for the proc->locks list
6165		   added by dlm_user_request, it may result in the lkb
6166		   being freed */
6167
6168		dlm_put_lkb(lkb);
6169	}
6170
6171	spin_lock(&ls->ls_clear_proc_locks);
6172
6173	/* in-progress unlocks */
6174	list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6175		list_del_init(&lkb->lkb_ownqueue);
6176		lkb->lkb_flags |= DLM_IFL_DEAD;
6177		dlm_put_lkb(lkb);
6178	}
6179
6180	list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6181		dlm_purge_lkb_callbacks(lkb);
 
6182		list_del_init(&lkb->lkb_cb_list);
6183		dlm_put_lkb(lkb);
6184	}
6185
6186	spin_unlock(&ls->ls_clear_proc_locks);
6187	dlm_unlock_recovery(ls);
6188}
6189
6190static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc)
6191{
6192	struct dlm_lkb *lkb, *safe;
6193
6194	while (1) {
6195		lkb = NULL;
6196		spin_lock(&proc->locks_spin);
6197		if (!list_empty(&proc->locks)) {
6198			lkb = list_entry(proc->locks.next, struct dlm_lkb,
6199					 lkb_ownqueue);
6200			list_del_init(&lkb->lkb_ownqueue);
6201		}
6202		spin_unlock(&proc->locks_spin);
6203
6204		if (!lkb)
6205			break;
6206
6207		lkb->lkb_flags |= DLM_IFL_DEAD;
6208		unlock_proc_lock(ls, lkb);
6209		dlm_put_lkb(lkb); /* ref from proc->locks list */
6210	}
6211
6212	spin_lock(&proc->locks_spin);
6213	list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) {
6214		list_del_init(&lkb->lkb_ownqueue);
6215		lkb->lkb_flags |= DLM_IFL_DEAD;
6216		dlm_put_lkb(lkb);
6217	}
6218	spin_unlock(&proc->locks_spin);
6219
6220	spin_lock(&proc->asts_spin);
6221	list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) {
6222		dlm_purge_lkb_callbacks(lkb);
 
6223		list_del_init(&lkb->lkb_cb_list);
6224		dlm_put_lkb(lkb);
6225	}
6226	spin_unlock(&proc->asts_spin);
6227}
6228
6229/* pid of 0 means purge all orphans */
6230
6231static void do_purge(struct dlm_ls *ls, int nodeid, int pid)
6232{
6233	struct dlm_lkb *lkb, *safe;
6234
6235	mutex_lock(&ls->ls_orphans_mutex);
6236	list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) {
6237		if (pid && lkb->lkb_ownpid != pid)
6238			continue;
6239		unlock_proc_lock(ls, lkb);
6240		list_del_init(&lkb->lkb_ownqueue);
6241		dlm_put_lkb(lkb);
6242	}
6243	mutex_unlock(&ls->ls_orphans_mutex);
6244}
6245
6246static int send_purge(struct dlm_ls *ls, int nodeid, int pid)
6247{
6248	struct dlm_message *ms;
6249	struct dlm_mhandle *mh;
6250	int error;
6251
6252	error = _create_message(ls, sizeof(struct dlm_message), nodeid,
6253				DLM_MSG_PURGE, &ms, &mh, GFP_NOFS);
6254	if (error)
6255		return error;
6256	ms->m_nodeid = cpu_to_le32(nodeid);
6257	ms->m_pid = cpu_to_le32(pid);
6258
6259	return send_message(mh, ms, NULL, 0);
6260}
6261
6262int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc,
6263		   int nodeid, int pid)
6264{
6265	int error = 0;
6266
6267	if (nodeid && (nodeid != dlm_our_nodeid())) {
6268		error = send_purge(ls, nodeid, pid);
6269	} else {
6270		dlm_lock_recovery(ls);
6271		if (pid == current->pid)
6272			purge_proc_locks(ls, proc);
6273		else
6274			do_purge(ls, nodeid, pid);
6275		dlm_unlock_recovery(ls);
6276	}
6277	return error;
6278}
6279
6280/* debug functionality */
6281int dlm_debug_add_lkb(struct dlm_ls *ls, uint32_t lkb_id, char *name, int len,
6282		      int lkb_nodeid, unsigned int lkb_flags, int lkb_status)
6283{
6284	struct dlm_lksb *lksb;
6285	struct dlm_lkb *lkb;
6286	struct dlm_rsb *r;
6287	int error;
6288
6289	/* we currently can't set a valid user lock */
6290	if (lkb_flags & DLM_IFL_USER)
6291		return -EOPNOTSUPP;
6292
6293	lksb = kzalloc(sizeof(*lksb), GFP_NOFS);
6294	if (!lksb)
6295		return -ENOMEM;
6296
6297	error = _create_lkb(ls, &lkb, lkb_id, lkb_id + 1);
6298	if (error) {
6299		kfree(lksb);
6300		return error;
6301	}
6302
6303	lkb->lkb_flags = lkb_flags;
6304	lkb->lkb_nodeid = lkb_nodeid;
6305	lkb->lkb_lksb = lksb;
6306	/* user specific pointer, just don't have it NULL for kernel locks */
6307	if (~lkb_flags & DLM_IFL_USER)
6308		lkb->lkb_astparam = (void *)0xDEADBEEF;
6309
6310	error = find_rsb(ls, name, len, 0, R_REQUEST, &r);
6311	if (error) {
6312		kfree(lksb);
6313		__put_lkb(ls, lkb);
6314		return error;
6315	}
6316
6317	lock_rsb(r);
6318	attach_lkb(r, lkb);
6319	add_lkb(r, lkb, lkb_status);
6320	unlock_rsb(r);
6321	put_rsb(r);
6322
6323	return 0;
6324}
6325
6326int dlm_debug_add_lkb_to_waiters(struct dlm_ls *ls, uint32_t lkb_id,
6327				 int mstype, int to_nodeid)
6328{
6329	struct dlm_lkb *lkb;
6330	int error;
6331
6332	error = find_lkb(ls, lkb_id, &lkb);
6333	if (error)
6334		return error;
6335
6336	error = add_to_waiters(lkb, mstype, to_nodeid);
6337	dlm_put_lkb(lkb);
6338	return error;
6339}
6340