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