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