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